Repository: istio/fortio (go)

Issue:
2*c connection leaks when using the UI to trigger run (regression?)
reloading many times while testing histogram boundary:


http://localhost:8080/fortio/?labels=Fortio&url=http%3A%2F%2Flocalhost%3A8080%2Fecho%3Fdelay%3D1ms%3A90%2C9ms%3A10&qps=100&t=&t=on&n=100&c=10&p=50%2C+75%2C+90%2C+99%2C+99.9&r=0.0000001&H=User-Agent%3A+istio%2Ffortio-0.7.2-pre&H=&H=&runner=http&save=on&load=Start


Starting at 100 qps with 10 thread(s) [gomax 4] : exactly 100, 10 calls each (total 100 + 0)
URL: localhost:8080/echo?delay=1ms:90,9ms:10
```
23:32:05 I uihandler.go:439> UI: GET /fortio/?labels=Fortio&url=http%3A%2F%2Flocalhost%3A8080%2Fecho%3Fdelay%3D1ms%3A90%2C9ms%3A10&qps=100&t=&t=on&n=100&c=10&p=50%2C+75%2C+90%2C+99%2C+99.9&r=0.0000001&H=User-Agent%3A+istio%2Ffortio-0.7.2-pre&H=&H=&runner=http&save=on&load=Start HTTP/1.1 [::1]:61957 ()
23:32:05 I uihandler.go:153> Starting http load request from [::1]:61957 for http://localhost:8080/echo?delay=1ms:90,9ms:10
23:32:05 I uihandler.go:209> New run id 13
23:32:05 I uihandler.go:227> Estimating fixed #call 100 duration to 1 seconds 100
23:32:05 I httprunner.go:72> Starting http test for http://localhost:8080/echo?delay=1ms:90,9ms:10 with 10 threads at 100.0 qps
23:32:05 E http.go:556> Unable to connect to {::1 8080 } : dial tcp [::1]:8080: socket: too many open files
23:32:05 E http.go:556> Unable to connect to {::1 8080 } : dial tcp [::1]:8080: socket: too many open files
23:32:05 E http.go:556> Unable to connect to {::1 8080 } : dial tcp [::1]:8080: socket: too many open files
23:32:05 E http.go:556> Unable to connect to {::1 8080 } : dial tcp [::1]:8080: socket: too many open files
23:32:05 E http.go:556> Unable to connect to {::1 8080 } : dial tcp [::1]:8080: socket: too many open files
23:32:05 http: Accept error: accept tcp [::]:8080: accept: too many open files; retrying in 5ms
23:32:05 E http.go:663> Read error &{{0xc420e35b00}} {::1 8080 } 0 : read tcp [::1]:62091->[::1]:8080: read: connection reset by peer
23:32:05 http: Accept error: accept tcp [::]:8080: accept: too many open files; retrying in 5ms
23:32:05 E http.go:663> Read error &{{0xc420e35b80}} {::1 8080 } 0 : read tcp [::1]:62092->[::1]:8080: read: connection reset by peer
23:32:05 http: Accept error: accept tcp [::]:8080: accept: too many open files; retrying in 5ms
23:32:05 E http.go:663> Read error &{{0xc4206a1980}} {::1 8080 } 0 : read tcp [::1]:62094->[::1]:8080: read: connection reset by peer
23:32:05 E http.go:556> Unable to connect to {::1 8080 } : dial tcp [::1]:8080: socket: too many open files
23:32:05 E http.go:556> Unable to connect to {::1 8080 } : dial tcp [::1]:8080: socket: too many open files
```

Pull request:
2*c connection/goroutine leak fix (std client/ui)
- [x] Phase 1 : repro with tests failing
- [x] Phase 2: code fix

Also verified manually with lsof

Fixes #166 


Patch:
diff --git a/README.md b/README.md
index 8e1ed86..71e348d 100644
--- a/README.md
+++ b/README.md
@@ -29,7 +29,7 @@ docker run istio/fortio load http://www.google.com/ # For a test run
 Or download the binary distribution, for instance:
 
 ```shell
-curl -L https://github.com/istio/fortio/releases/download/v0.7.1/fortio-linux_x64-0.7.1.tgz \
+curl -L https://github.com/istio/fortio/releases/download/v0.7.2/fortio-linux_x64-0.7.2.tgz \
  | sudo tar -C / -xvzpf -
 ```
 
@@ -42,7 +42,7 @@ You can get a preview of the reporting/graphing UI at https://fortio.istio.io/
 Fortio can be an http or grpc load generator, gathering statistics using the `load` subcommand, or start simple http and grpc ping servers, as well as a basic web UI, result graphing and https redirector, with the `server` command or issue grpc ping messages using the `grpcping` command. It can also fetch a single URL's for debugging when using the `curl` command (or the `-curl` flag to the load command). You can run just the redirector with `redirect`. Lastly if you saved JSON results (using the web UI or directly from the command line), you can browse and graph those results using the `report` command.
 
 ```
-Φορτίο 0.7.1 usage:
+Φορτίο 0.7.2 usage:
 	fortio command [flags] target
 where command is one of: load (load testing), server (starts grpc ping and
 http echo/ui/redirect servers), grpcping (grpc client), report (report only UI
@@ -153,8 +153,8 @@ $ fortio server &
 Https redirector running on :8081
 UI starting - visit:
 http://localhost:8080/fortio/   (or any host/ip reachable on this server)
-Fortio 0.7.1 grpc ping server listening on port :8079
-Fortio 0.7.1 echo server listening on port :8080
+Fortio 0.7.2 grpc ping server listening on port :8079
+Fortio 0.7.2 echo server listening on port :8080
 ```
 
 * By default, Fortio's web/echo servers listen on port 8080 on all interfaces.
@@ -164,8 +164,8 @@ $ fortio server -http-port 10.10.10.10:8088
 UI starting - visit:
 http://10.10.10.10:8088/fortio/
 Https redirector running on :8081
-Fortio 0.7.1 grpc ping server listening on port :8079
-Fortio 0.7.1 echo server listening on port 10.10.10.10:8088
+Fortio 0.7.2 grpc ping server listening on port :8079
+Fortio 0.7.2 echo server listening on port 10.10.10.10:8088
 ```
 * Simple grpc ping:
 ```
@@ -235,14 +235,14 @@ Content-Type: text/plain; charset=UTF-8
 Date: Mon, 08 Jan 2018 22:26:26 GMT
 Content-Length: 230
 
-Φορτίο version 0.7.1 echo debug server up for 39s on ldemailly-macbookpro - request from [::1]:65055
+Φορτίο version 0.7.2 echo debug server up for 39s on ldemailly-macbookpro - request from [::1]:65055
 
 GET /debug HTTP/1.1
 
 headers:
 
 Host: localhost:8080
-User-Agent: istio/fortio-0.7.1
+User-Agent: istio/fortio-0.7.2
 Foo: Bar
 
 body:
diff --git a/fhttp/http.go b/fhttp/http.go
index 988c85a..b8145ae 100644
--- a/fhttp/http.go
+++ b/fhttp/http.go
@@ -47,6 +47,8 @@ type Fetcher interface {
 	// Fetch returns http code, data, offset of body (for client which returns
 	// headers)
 	Fetch() (int, []byte, int)
+	// Close() cleans up connections and state - must be paired with NewClient calls.
+	Close()
 }
 
 var (
@@ -206,9 +208,26 @@ func newHTTPRequest(o *HTTPOptions) *http.Request {
 // Client object for making repeated requests of the same URL using the same
 // http client (net/http)
 type Client struct {
-	url    string
-	req    *http.Request
-	client *http.Client
+	url       string
+	req       *http.Request
+	client    *http.Client
+	transport *http.Transport
+}
+
+// Close cleans up any resources used by NewStdClient
+func (c *Client) Close() {
+	log.Debugf("Close() on %+v", c)
+	if c.req != nil {
+		if c.req.Body != nil {
+			if err := c.req.Body.Close(); err != nil {
+				log.Warnf("Error closing std client body: %v", err)
+			}
+		}
+		c.req = nil
+	}
+	if c.transport != nil {
+		c.transport.CloseIdleConnections()
+	}
 }
 
 // ChangeURL only for standard client, allows fetching a different URL
@@ -257,7 +276,7 @@ func NewClient(o *HTTPOptions) Fetcher {
 	if o.DisableFastClient {
 		return NewStdClient(o)
 	}
-	return NewBasicClient(o)
+	return NewFastClient(o)
 }
 
 // NewStdClient creates a client object that wraps the net/http standard client.
@@ -298,12 +317,13 @@ func NewStdClient(o *HTTPOptions) *Client {
 				return http.ErrUseLastResponse
 			},
 		},
+		&tr,
 	}
 	return &client
 }
 
-// BasicClient is a fast, lockfree single purpose http 1.0/1.1 client.
-type BasicClient struct {
+// FastClient is a fast, lockfree single purpose http 1.0/1.1 client.
+type FastClient struct {
 	buffer       []byte
 	req          []byte
 	dest         net.TCPAddr
@@ -323,10 +343,21 @@ type BasicClient struct {
 	reqTimeout   time.Duration
 }
 
-// NewBasicClient makes a basic, efficient http 1.0/1.1 client.
+// Close cleans up any resources used by FastClient
+func (c *FastClient) Close() {
+	log.Debugf("Closing %+v", c)
+	if c.socket != nil {
+		if err := c.socket.Close(); err != nil {
+			log.Warnf("Error closing fast client's socket: %v", err)
+		}
+		c.socket = nil
+	}
+}
+
+// NewFastClient makes a basic, efficient http 1.0/1.1 client.
 // This function itself doesn't need to be super efficient as it is created at
 // the beginning and then reused many times.
-func NewBasicClient(o *HTTPOptions) Fetcher {
+func NewFastClient(o *HTTPOptions) Fetcher {
 	proto := "1.1"
 	if o.HTTP10 {
 		proto = "1.0"
@@ -342,7 +373,7 @@ func NewBasicClient(o *HTTPOptions) Fetcher {
 		return nil
 	}
 	// note: Host includes the port
-	bc := BasicClient{url: o.URL, host: url.Host, hostname: url.Hostname(), port: url.Port(),
+	bc := FastClient{url: o.URL, host: url.Host, hostname: url.Hostname(), port: url.Port(),
 		http10: o.HTTP10, halfClose: o.AllowHalfClose}
 	bc.buffer = make([]byte, BufferSizeKb*1024)
 	if bc.port == "" {
@@ -545,12 +576,12 @@ func ParseChunkSize(inp []byte) (int, int) {
 }
 
 // return the result from the state.
-func (c *BasicClient) returnRes() (int, []byte, int) {
+func (c *FastClient) returnRes() (int, []byte, int) {
 	return c.code, c.buffer[:c.size], c.headerLen
 }
 
 // connect to destination.
-func (c *BasicClient) connect() *net.TCPConn {
+func (c *FastClient) connect() *net.TCPConn {
 	socket, err := net.DialTCP("tcp", nil, &c.dest)
 	if err != nil {
 		log.Errf("Unable to connect to %v : %v", c.dest, err)
@@ -570,7 +601,7 @@ func (c *BasicClient) connect() *net.TCPConn {
 }
 
 // Fetch fetches the url content. Returns http code, data, offset of body.
-func (c *BasicClient) Fetch() (int, []byte, int) {
+func (c *FastClient) Fetch() (int, []byte, int) {
 	c.code = -1
 	c.size = 0
 	c.headerLen = 0
@@ -637,7 +668,7 @@ func DebugSummary(buf []byte, max int) string {
 
 // Response reading:
 // TODO: refactor - unwiedly/ugly atm
-func (c *BasicClient) readResponse(conn *net.TCPConn) {
+func (c *FastClient) readResponse(conn *net.TCPConn) {
 	max := len(c.buffer)
 	parsedHeaders := false
 	// TODO: safer to start with -1 and fix ok for http 1.0
diff --git a/fhttp/httprunner.go b/fhttp/httprunner.go
index fc85db7..ef67489 100644
--- a/fhttp/httprunner.go
+++ b/fhttp/httprunner.go
@@ -125,10 +125,11 @@ func RunHTTPTest(o *HTTPRunnerOptions) (*HTTPRunnerResults, error) {
 		fm.Close()                 // nolint:gas,errcheck
 		fmt.Fprintf(out, "Wrote profile data to %s.{cpu|mem}\n", o.Profiler)
 	}
-	// Numthreads may have reduced
-	numThreads = r.Options().NumThreads
+	// Numthreads may have reduced but it should be ok to accumulate 0s from
+	// unused ones. We also must cleanup all the created clients.
 	keys := []int{}
 	for i := 0; i < numThreads; i++ {
+		httpstate[i].client.Close()
 		// Q: is there some copying each time stats[i] is used?
 		for k := range httpstate[i].RetCodes {
 			if _, exists := total.RetCodes[k]; !exists {
diff --git a/ui/uihandler.go b/ui/uihandler.go
index e8b76e8..8494ad8 100644
--- a/ui/uihandler.go
+++ b/ui/uihandler.go
@@ -601,6 +601,7 @@ func FetcherHandler(w http.ResponseWriter, r *http.Request) {
 	if err != nil {
 		log.Errf("Error writing fetched data to %v: %v", r.RemoteAddr, err)
 	}
+	client.Close()
 }
 
 func onBehalfOf(o *fhttp.HTTPOptions, r *http.Request) {
@@ -678,6 +679,7 @@ func SyncHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	code, data, _ := client.Fetch()
+	defer client.Close()
 	if code != http.StatusOK {
 		w.Write([]byte(fmt.Sprintf("http error, code %d<script>setPB(1,1)</script></body></html>\n", code))) // nolint: gas, errcheck
 		w.WriteHeader(424 /*Failed Dependency*/)


Test patch:
diff --git a/Webtest.sh b/Webtest.sh
index 46d5d9e..c860c48 100755
--- a/Webtest.sh
+++ b/Webtest.sh
@@ -3,9 +3,9 @@ set -x
 # Check we can build the image
 make docker-internal TAG=webtest || exit 1
 FORTIO_UI_PREFIX=/newprefix/ # test the non default prefix (not /fortio/)
-#LOGLEVEL=debug
-LOGLEVEL=info
-DOCKERID=$(docker run -d --name fortio_server istio/fortio:webtest server -ui-path $FORTIO_UI_PREFIX -loglevel $LOGLEVEL)
+FILE_LIMIT=20 # must be low to detect leaks
+LOGLEVEL=info # change to debug to debug
+DOCKERID=$(docker run -d --ulimit nofile=$FILE_LIMIT --name fortio_server istio/fortio:webtest server -ui-path $FORTIO_UI_PREFIX -loglevel $LOGLEVEL)
 function cleanup {
   docker stop $DOCKERID
   docker rm fortio_server
@@ -23,6 +23,8 @@ $CURL https://istio.io/robots.txt
 $CURL ${BASE_FORTIO}browse
 # Check we can connect, and run a http QPS test against ourselves through fetch
 $CURL "${BASE_FORTIO}fetch/localhost:8080$FORTIO_UI_PREFIX?url=http://localhost:8080/debug&load=Start&qps=-1&json=on" | grep ActualQPS
+# Check we can do it twice despite ulimit - check we get all 200s (exactly 80 of them (default is 8 connections->16 fds + a few))
+$CURL "${BASE_FORTIO}fetch/localhost:8080$FORTIO_UI_PREFIX?url=http://localhost:8080/debug&load=Start&n=80&qps=-1&json=on" | grep '"200": 80'
 # Check we can connect, and run a grpc QPS test against ourselves through fetch
 $CURL "${BASE_FORTIO}fetch/localhost:8080$FORTIO_UI_PREFIX?url=localhost:8079&load=Start&qps=-1&json=on&n=100&runner=grpc" | grep '"1": 100'
 # Check we get the logo (need to remove the CR from raw headers)
diff --git a/fhttp/httprunner_test.go b/fhttp/httprunner_test.go
index 3193f90..1d15e5a 100644
--- a/fhttp/httprunner_test.go
+++ b/fhttp/httprunner_test.go
@@ -22,6 +22,7 @@ package fhttp
 import (
 	"fmt"
 	"net/http"
+	"runtime"
 	"strings"
 	"testing"
 	"time"
@@ -60,7 +61,7 @@ func TestHTTPRunner(t *testing.T) {
 		URL: opts.URL,
 	}
 	o1 := rawOpts
-	if r, _, _ := NewBasicClient(&o1).Fetch(); r != http.StatusOK {
+	if r, _, _ := NewFastClient(&o1).Fetch(); r != http.StatusOK {
 		t.Errorf("Fast Client with raw option should still work with warning in logs")
 	}
 	o1 = rawOpts
@@ -72,6 +73,57 @@ func TestHTTPRunner(t *testing.T) {
 	}
 }
 
+func testHTTPNotLeaking(t *testing.T, opts *HTTPRunnerOptions) {
+	ngBefore1 := runtime.NumGoroutine()
+	t.Logf("Number go routine before test %d", ngBefore1)
+	port, mux := DynamicHTTPServer(false)
+	mux.HandleFunc("/echo100", EchoHandler)
+	url := fmt.Sprintf("http://localhost:%d/echo100", port)
+	numCalls := 100
+	opts.NumThreads = numCalls / 2 // make 2 calls per thread
+	opts.Exactly = int64(numCalls)
+	opts.QPS = float64(numCalls) / 2 // take 1 second
+	opts.URL = url
+	// Warm up round 1
+	res, err := RunHTTPTest(opts)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	httpOk := res.RetCodes[http.StatusOK]
+	if opts.Exactly != httpOk {
+		t.Errorf("Run1: Mismatch between requested calls %d and ok %v", numCalls, res.RetCodes)
+	}
+	ngBefore2 := runtime.NumGoroutine()
+	t.Logf("Number of go routine after warm up / before 2nd test %d", ngBefore2)
+	// 2nd run, should be stable number of go routines after first, not keep growing:
+	res, err = RunHTTPTest(opts)
+	// it takes a while for the connections to close with std client (!) why isn't CloseIdleConnections() synchronous
+	runtime.GC()
+	ngAfter := runtime.NumGoroutine()
+	t.Logf("Number of go routine after 2nd test %d", ngAfter)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	httpOk = res.RetCodes[http.StatusOK]
+	if opts.Exactly != httpOk {
+		t.Errorf("Run2: Mismatch between requested calls %d and ok %v", numCalls, res.RetCodes)
+	}
+	// allow for ~5 goroutine variance, as we use 50 if we leak it will show (was failing before #167)
+	if ngAfter > ngBefore2+5 {
+		t.Errorf("Goroutines after test %d, expected it to stay near %d", ngAfter, ngBefore2)
+	}
+}
+
+func TestHttpNotLeakingFastClient(t *testing.T) {
+	testHTTPNotLeaking(t, &HTTPRunnerOptions{})
+}
+
+func TestHttpNotLeakingStdClient(t *testing.T) {
+	testHTTPNotLeaking(t, &HTTPRunnerOptions{HTTPOptions: HTTPOptions{DisableFastClient: true}})
+}
+
 func TestHTTPRunnerClientRace(t *testing.T) {
 	port, mux := DynamicHTTPServer(false)
 	mux.HandleFunc("/echo1/", EchoHandler)