Repository: fortio/fortio (go)

Issue:
option to client side force not reusing a connection more than X times
option to client side force not reusing a connection more than X times
default would remain to reuse connection as much as possible (until an error or server close) but we can add a
`-max-connection-reuse n`

One issue is maybe we would want different numbers per thread so they don't do it all at the same time (could be offset by % threadid though that is yet another option (so many/too many)

combined with #532 and #531 we would simulate more easily "new clients"

Pull request:
Max connection reuse
### Feature
Add flag `max-connection-reuse` to force create new connection after a thread reached the max connection reuse. 
### Description
The flag value is a string `min:max` to define the range of max connection reuse thresholds that thread will randomly pick a value within that range. This range will avoid all thread create new connection at the same time.
### Example
```
❯ go run . load -loglevel verbose  -max-connection-reuse 10:20 -c 2 -n 50  https://fortio.org/
...
Starting at 8 qps with 2 thread(s) [gomax 10] : exactly 50, 25 calls each (total 50 + 0)
...
15:53:26 V http_client.go:727> [0] Thread reach the threshold for max connection reuse of 17, force create new connection
15:53:27 V http_client.go:727> [1] Thread reach the threshold for max connection reuse of 12, force create new connection
...
All done 50 calls (plus 0 warmup) 337.438 ms avg, 5.9 qps
```

Close #533

Patch:
diff --git a/.golangci.yml b/.golangci.yml
index 66fe496..894f0ac 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -342,6 +342,7 @@ linters-settings:
 linters:
   enable:
     - unused
+    - errorlint
 #    - megacheck
 #    - govet
   disable:
diff --git a/README.md b/README.md
index f12e7a0..0613b43 100644
--- a/README.md
+++ b/README.md
@@ -51,13 +51,13 @@ You can install from source:
 The [releases](https://github.com/fortio/fortio/releases) page has binaries for many OS/architecture combinations (see assets).
 
 ```shell
-curl -L https://github.com/fortio/fortio/releases/download/v1.33.0/fortio-linux_amd64-1.33.0.tgz \
+curl -L https://github.com/fortio/fortio/releases/download/v1.34.0/fortio-linux_amd64-1.34.0.tgz \
  | sudo tar -C / -xvzpf -
 # or the debian package
-wget https://github.com/fortio/fortio/releases/download/v1.33.0/fortio_1.33.0_amd64.deb
-dpkg -i fortio_1.33.0_amd64.deb
+wget https://github.com/fortio/fortio/releases/download/v1.34.0/fortio_1.34.0_amd64.deb
+dpkg -i fortio_1.34.0_amd64.deb
 # or the rpm
-rpm -i https://github.com/fortio/fortio/releases/download/v1.33.0/fortio-1.33.0-1.x86_64.rpm
+rpm -i https://github.com/fortio/fortio/releases/download/v1.34.0/fortio-1.34.0-1.x86_64.rpm
 # and more, see assets in release page
 ```
 
@@ -67,7 +67,7 @@ On a MacOS you can also install Fortio using [Homebrew](https://brew.sh/):
 brew install fortio
 ```
 
-On Windows, download https://github.com/fortio/fortio/releases/download/v1.33.0/fortio_win_1.33.0.zip and extract `fortio.exe` to any location, then using the Windows Command Prompt:
+On Windows, download https://github.com/fortio/fortio/releases/download/v1.34.0/fortio_win_1.34.0.zip and extract `fortio.exe` to any location, then using the Windows Command Prompt:
 ```
 fortio.exe server
 ```
@@ -115,7 +115,7 @@ Full list of command line flags (`fortio help`):
 <details>
 <!-- use release/updateFlags.sh to update this section -->
 <pre>
-Φορτίο 1.33.0 usage:
+Φορτίο 1.34.0 usage:
 where command is one of: load (load testing), server (starts ui, http-echo,
  redirect, proxies, tcp-echo and grpc ping servers), tcp-echo (only the tcp-echo
  server), report (report only UI server), redirect (only the redirect server),
@@ -160,6 +160,9 @@ connections, if empty, use https:// prefix for standard internet/system CAs
   -config path
         Config directory path to watch for changes of dynamic flags (empty for
 no watch)
+  -connection-reuse-range min:max
+        Range min:max for the max number of connections to reuse for each thread, 
+default to unlimited. e.g. 10:30 means randomly choose a max connection reuse threshold between 10 and 30 requests.
   -content-type string
         Sets http content type. Setting this value switches the request method
 from GET to POST.
diff --git a/bincommon/commonflags.go b/bincommon/commonflags.go
index 589c1e1..8f44449 100644
--- a/bincommon/commonflags.go
+++ b/bincommon/commonflags.go
@@ -27,6 +27,7 @@ import (
 	"reflect"
 	"strings"
 
+	"fortio.org/fortio/dflag"
 	"fortio.org/fortio/fhttp"
 	"fortio.org/fortio/fnet"
 	"fortio.org/fortio/log"
@@ -106,6 +107,11 @@ var (
 		"http(s) runner warmup done in parallel instead of sequentially. When set, restores pre 1.21 behavior")
 	curlHeadersStdout = flag.Bool("curl-stdout-headers", false,
 		"Restore pre 1.22 behavior where http headers of the fast client are output to stdout in curl mode. now stderr by default.")
+	// ConnectionReuseRange Dynamic string flag to set the max connection reuse range.
+	ConnectionReuseRange = dflag.DynString(flag.CommandLine, "connection-reuse-range", "",
+		"Range `min:max` for the max number of connections to reuse for each thread, default to unlimited. "+
+			"e.g. 10:30 means randomly choose a max connection reuse threshold between 10 and 30 requests.").
+		WithValidator(ConnectionReuseRangeValidator(&httpOpts))
 )
 
 // SharedMain is the common part of main from fortio_main and fcurl.
@@ -170,6 +176,18 @@ func TLSInsecure() bool {
 	return TLSInsecure
 }
 
+// ConnectionReuseRangeValidator returns a validator function that checks if the connection reuse range is valid
+// and set in httpOpts.
+func ConnectionReuseRangeValidator(httpOpts *fhttp.HTTPOptions) func(string) error {
+	return func(value string) error {
+		if err := httpOpts.ValidateAndSetConnectionReuseRange(value); err != nil {
+			return err
+		}
+
+		return nil
+	}
+}
+
 // SharedHTTPOptions is the flag->httpoptions transfer code shared between
 // fortio_main and fcurl.
 func SharedHTTPOptions() *fhttp.HTTPOptions {
diff --git a/fhttp/http_client.go b/fhttp/http_client.go
index 893b392..4e17457 100644
--- a/fhttp/http_client.go
+++ b/fhttp/http_client.go
@@ -184,6 +184,7 @@ type HTTPOptions struct {
 	LogErrors        bool          // whether to log non 2xx code as they occur or not
 	ID               int           // id to use for logging (thread id when used as a runner)
 	SequentialWarmup bool          // whether to do http(s) runs warmup sequentially or in parallel (new default is //)
+	ConnReuseRange   [2]int        // range of max number of connection to reuse for each thread.
 }
 
 // ResetHeaders resets all the headers, including the User-Agent: one (and the Host: logical special header).
@@ -277,6 +278,39 @@ func (h *HTTPOptions) AddAndValidateExtraHeader(hdr string) error {
 	return nil
 }
 
+func (h *HTTPOptions) ValidateAndSetConnectionReuseRange(inp string) error {
+	if inp == "" {
+		return nil
+	}
+
+	reuseRangeString := strings.Split(inp, ":")
+	var reuseRangeInt []int
+
+	if len(reuseRangeString) > 2 {
+		return fmt.Errorf("more than two integers were provided in the connection reuse range")
+	}
+
+	for _, input := range reuseRangeString {
+		if val, err := strconv.Atoi(input); err != nil {
+			return fmt.Errorf("invalid value for connection reuse range, err: %w", err)
+		} else {
+			reuseRangeInt = append(reuseRangeInt, val)
+		}
+	}
+
+	if len(reuseRangeInt) == 1 {
+		h.ConnReuseRange = [2]int{reuseRangeInt[0], reuseRangeInt[0]}
+	} else {
+		if reuseRangeInt[0] < reuseRangeInt[1] {
+			h.ConnReuseRange = [2]int{reuseRangeInt[0], reuseRangeInt[1]}
+		} else {
+			h.ConnReuseRange = [2]int{reuseRangeInt[1], reuseRangeInt[0]}
+		}
+	}
+
+	return nil
+}
+
 // newHttpRequest makes a new http GET request for url with User-Agent.
 func newHTTPRequest(o *HTTPOptions) (*http.Request, error) {
 	method := o.Method()
@@ -545,6 +579,10 @@ type FastClient struct {
 	id           int
 	https        bool
 	tlsConfig    *tls.Config
+	// range of connection reuse threshold that current thread will choose from
+	connReuseRange [2]int
+	connReuse      int
+	reuseCount     int
 }
 
 // GetIPAddress get the hostname and ip address that DNS resolved to when using fast client.
@@ -598,11 +636,18 @@ func NewFastClient(o *HTTPOptions) (Fetcher, error) {
 		log.Errf("[%d] Bad url %q : %v", o.ID, urlString, err)
 		return nil, err
 	}
+
+	// Randomly assign a max connection reuse threshold to this thread.
+	var connReuse int
+	if o.ConnReuseRange != [2]int{0, 0} {
+		connReuse = generateReuseThreshold(o.ConnReuseRange[0], o.ConnReuseRange[1])
+	}
+
 	// note: Host includes the port
 	bc := FastClient{
 		url: o.URL, host: url.Host, hostname: url.Hostname(), port: url.Port(),
 		http10: o.HTTP10, halfClose: o.AllowHalfClose, logErrors: o.LogErrors, id: o.ID,
-		https: o.https,
+		https: o.https, connReuseRange: o.ConnReuseRange, connReuse: connReuse,
 	}
 	if o.https {
 		bc.tlsConfig, err = o.TLSOptions.TLSClientConfig()
@@ -722,13 +767,21 @@ func (c *FastClient) Fetch() (int, []byte, int) {
 	c.headerLen = 0
 	// Connect or reuse existing socket:
 	conn := c.socket
-	reuse := (conn != nil)
-	if !reuse {
+	canReuse := conn != nil
+	if c.reachedReuseThreshold() {
+		c.connReuse = generateReuseThreshold(c.connReuseRange[0], c.connReuseRange[1])
+		log.LogVf("[%d] Thread reach the threshold for max connection canReuse of %d, force create new connection",
+			c.id, c.connReuse)
+	}
+
+	if conn == nil {
 		conn = c.connect()
+		c.reuseCount = 1
 		if conn == nil {
 			return c.returnRes()
 		}
 	} else {
+		c.reuseCount++
 		log.Debugf("[%d] Reusing socket %v", c.id, c.dest)
 	}
 	c.socket = nil // because of error returns and single retry
@@ -742,7 +795,7 @@ func (c *FastClient) Fetch() (int, []byte, int) {
 	}
 	n, err := conn.Write(req)
 	if err != nil || conErr != nil {
-		if reuse {
+		if canReuse {
 			// it's ok for the (idle) socket to die once, auto reconnect:
 			log.Infof("[%d] Closing dead socket %v (%v)", c.id, c.dest, err)
 			conn.Close()
@@ -769,7 +822,7 @@ func (c *FastClient) Fetch() (int, []byte, int) {
 		}
 	}
 	// Read the response:
-	c.readResponse(conn, reuse)
+	c.readResponse(conn, canReuse)
 	if c.code == RetryOnce {
 		// Special "eof on reused socket" code
 		return c.Fetch() // recurse once
@@ -963,7 +1016,7 @@ func (c *FastClient) readResponse(conn net.Conn, reusedSocket bool) {
 		}
 	} // end of big for loop
 	// Figure out whether to keep or close the socket:
-	if keepAlive && codeIsOK(c.code) {
+	if keepAlive && codeIsOK(c.code) && !c.reachedReuseThreshold() {
 		c.socket = conn // keep the open socket
 	} else {
 		if err := conn.Close(); err != nil {
@@ -975,7 +1028,27 @@ func (c *FastClient) readResponse(conn net.Conn, reusedSocket bool) {
 	}
 }
 
+// Check if current thread reached the connection reuse threshold.
+func (c *FastClient) reachedReuseThreshold() bool {
+	if c.connReuse != 0 && c.reuseCount >= c.connReuse {
+		log.LogVf("[%d] Thread reach the threshold for max connection reuse of %d, force create new connection",
+			c.id, c.connReuse)
+		return true
+	}
+
+	return false
+}
+
 func generateUUID() string {
 	// We use math random instead of crypto random generator due to performance.
 	return uuid.Must(uuid.NewRandomFromReader(rander)).String()
 }
+
+// Generate reuse threshold based on the min and max value in the flag.
+func generateReuseThreshold(min int, max int) int {
+	if min == max {
+		return min
+	}
+
+	return min + rand.Intn(max-min+1)
+}
diff --git a/ui/restHandler.go b/ui/restHandler.go
index ff680df..37eb60f 100644
--- a/ui/restHandler.go
+++ b/ui/restHandler.go
@@ -11,6 +11,7 @@ import (
 	"strings"
 	"time"
 
+	"fortio.org/fortio/bincommon"
 	"fortio.org/fortio/fgrpc"
 	"fortio.org/fortio/fhttp"
 	"fortio.org/fortio/log"
@@ -192,6 +193,14 @@ func RESTRunHandler(w http.ResponseWriter, r *http.Request) { // nolint: funlen
 	httpopts.SequentialWarmup = sequentialWarmup
 	httpopts.Insecure = httpsInsecure
 	httpopts.Resolve = resolve
+	// Set the connection reuse range.
+	err = bincommon.ConnectionReuseRange.
+		WithValidator(bincommon.ConnectionReuseRangeValidator(httpopts)).
+		Set(FormValue(r, jd, "connection-reuse-range"))
+	if err != nil {
+		log.Errf("Fail to validate connection reuse range flag, err: %v", err)
+	}
+
 	if len(payload) > 0 {
 		httpopts.Payload = []byte(payload)
 	}
diff --git a/ui/templates/main.html b/ui/templates/main.html
index 34a9c6d..59c4f21 100644
--- a/ui/templates/main.html
+++ b/ui/templates/main.html
@@ -46,6 +46,9 @@
     or run until interrupted:<input type="checkbox" name="t" onchange="toggleDuration(this)" />
     or run for exactly <input type="text" name="n" size="6" value="" /> calls. <br />
     Threads/Simultaneous connections: <input type="text" name="c" size="6" value="8" /> <br />
+    Connection reuse range: Min <input type="text" name="connection-reuse-range-min" size="6" value="" />
+    Max <input type="text" name="connection-reuse-range-max" size="6" value="" />
+    or Single value: <input type="text" name="connection-reuse-range-value" size="6" value="" /> <br />
     Uniform:<input type="checkbox" name="uniform" /> or Jitter:<input type="checkbox" name="jitter" /> &nbsp;&nbsp;
     No Catch-Up (qps is a ceiling): <input type="checkbox" name="nocatchup" /><br />
     Percentiles: <input type="text" name="p" size="20" value="50, 75, 90, 99, 99.9" /> <br />
diff --git a/ui/uihandler.go b/ui/uihandler.go
index 8b18fdd..4a7d694 100644
--- a/ui/uihandler.go
+++ b/ui/uihandler.go
@@ -38,6 +38,7 @@ import (
 	"sync"
 	"time"
 
+	"fortio.org/fortio/bincommon"
 	"fortio.org/fortio/dflag/endpoint"
 	"fortio.org/fortio/fgrpc"
 	"fortio.org/fortio/fhttp"
@@ -146,6 +147,10 @@ func Handler(w http.ResponseWriter, r *http.Request) {
 	percList, _ := stats.ParsePercentiles(r.FormValue("p"))
 	qps, _ := strconv.ParseFloat(r.FormValue("qps"), 64)
 	durStr := r.FormValue("t")
+	connectionReuseRange := parseConnectionReuseRange(
+		r.FormValue("connection-reuse-range-min"),
+		r.FormValue("connection-reuse-range-max"),
+		r.FormValue("connection-reuse-range-value"))
 	jitter := (r.FormValue("jitter") == "on")
 	uniform := (r.FormValue("uniform") == "on")
 	nocatchup := (r.FormValue("nocatchup") == "on")
@@ -216,6 +221,14 @@ func Handler(w http.ResponseWriter, r *http.Request) {
 	httpopts.SequentialWarmup = sequentialWarmup
 	httpopts.Insecure = httpsInsecure
 	httpopts.Resolve = resolve
+	// Set the connection reuse range.
+	err := bincommon.ConnectionReuseRange.
+		WithValidator(bincommon.ConnectionReuseRangeValidator(httpopts)).
+		Set(connectionReuseRange)
+	if err != nil {
+		log.Errf("Fail to validate connection reuse range flag, err: %v", err)
+	}
+
 	if len(payload) > 0 {
 		httpopts.Payload = []byte(payload)
 	}
@@ -1019,3 +1032,13 @@ func Report(baseurl, port, datadir string) bool {
 	mux.Handle(uiPath+"data/", LogAndFilterDataRequest(http.StripPrefix(uiPath+"data", fsd)))
 	return true
 }
+
+func parseConnectionReuseRange(min string, max string, value string) string {
+	if min != "" && max != "" {
+		return fmt.Sprintf("%s:%s", min, max)
+	} else if value != "" {
+		return value
+	}
+
+	return ""
+}


Test patch:
diff --git a/fhttp/httprunner_test.go b/fhttp/httprunner_test.go
index da3631a..100b23e 100644
--- a/fhttp/httprunner_test.go
+++ b/fhttp/httprunner_test.go
@@ -20,6 +20,7 @@ import (
 	"bytes"
 	"compress/gzip"
 	"fmt"
+	"math"
 	"net/http"
 	"os"
 	"path"
@@ -471,6 +472,108 @@ func TestAbortOn(t *testing.T) {
 	}
 }
 
+func TestConnectionReuseRange(t *testing.T) {
+	mux, addr := DynamicHTTPServer(false)
+	mux.HandleFunc("/foo/", EchoHandler)
+	url := fmt.Sprintf("http://localhost:%d/foo/", addr.Port)
+	opts := HTTPRunnerOptions{}
+	opts.Init(url)
+	opts.QPS = 50
+	opts.URL = url
+	opts.NumThreads = 1
+	opts.Exactly = 10
+
+	// Test empty reuse range.
+	res, err := RunHTTPTest(&opts)
+	if err != nil {
+		t.Error(err)
+	}
+
+	if res.SocketCount != 1 {
+		t.Errorf("Expected only 1 socket should be used when the max connection reuse flag is not set.")
+	}
+
+	// Test connection reuse range from 1 to 10.
+	for i := 1; i <= 10; i++ {
+		opts.ConnReuseRange = [2]int{i, i}
+		expectedSocketReuse := math.Ceil(float64(opts.Exactly) / float64(opts.ConnReuseRange[0]))
+		res, err := RunHTTPTest(&opts)
+		if err != nil {
+			t.Error(err)
+		}
+
+		if res.SocketCount != (int)(expectedSocketReuse) {
+			t.Errorf("Expecting %f socket to be used, got %d", expectedSocketReuse, res.SocketCount)
+		}
+	}
+
+	// Test when connection reuse range min != max.
+	// The actual socket count should always be 2 as the connection reuse range varies between 5 and 9.
+	expectedSocketReuse := 2
+	opts.ConnReuseRange = [2]int{5, 9}
+	// Check a few times that despite the range and random 2-9 we still always get 2 connections
+	for i := 0; i < 5; i++ {
+		res, err := RunHTTPTest(&opts)
+		if err != nil {
+			t.Error(err)
+		}
+
+		if res.SocketCount != expectedSocketReuse {
+			t.Errorf("Expecting %d socket to be used, got %d", expectedSocketReuse, res.SocketCount)
+		}
+	}
+}
+
+func TestValidateConnectionReuse(t *testing.T) {
+	httpOpts := HTTPOptions{}
+
+	err := httpOpts.ValidateAndSetConnectionReuseRange("1:2:3:4")
+	if err == nil {
+		t.Errorf("Shoud fail when more than two values are provided for connection reuse range.")
+	}
+
+	err = httpOpts.ValidateAndSetConnectionReuseRange("foo")
+	if err == nil {
+		t.Errorf("Shoud fail when non integer value is provided for connection reuse range.")
+	}
+
+	err = httpOpts.ValidateAndSetConnectionReuseRange("")
+	if err != nil {
+		t.Errorf("Expect no error when no value is privided, got err: %v.", err)
+	}
+
+	if httpOpts.ConnReuseRange != [2]int{0, 0} {
+		t.Errorf("Expect the connection reuse range to be [0, 0], got : %v.", httpOpts.ConnReuseRange)
+	}
+
+	err = httpOpts.ValidateAndSetConnectionReuseRange("10")
+	if err != nil {
+		t.Errorf("Expect no error when single value is privided, got err: %v.", err)
+	}
+
+	if httpOpts.ConnReuseRange != [2]int{10, 10} {
+		t.Errorf("Expect the connection reuse range to be [10, 10], got : %v.", httpOpts.ConnReuseRange)
+	}
+
+	err = httpOpts.ValidateAndSetConnectionReuseRange("20:10")
+	if err != nil {
+		t.Errorf("Expect no error when two values are privided, got err: %v.", err)
+	}
+
+	if httpOpts.ConnReuseRange != [2]int{10, 20} {
+		t.Errorf("Expect the connection reuse range to be [10, 20], got : %v.", httpOpts.ConnReuseRange)
+	}
+
+	err = httpOpts.ValidateAndSetConnectionReuseRange("10:20")
+	if err != nil {
+		t.Errorf("Expect no error when two values are privided, got err: %v", err)
+	}
+
+	if httpOpts.ConnReuseRange != [2]int{10, 20} {
+		t.Errorf("Expect the connection reuse range to be [10, 20], got : %v.", httpOpts.ConnReuseRange)
+	}
+}
+
 func getIPUsageCount(ipCountMap map[string]int) (count int) {
 	for _, v := range ipCountMap {
 		count += v