add configurable upstream connection limits

Add new --upstream.max-idle-conns and --upstream.max-conns options
to configure the HTTP transport connection pool settings.

- max-idle-conns: max idle connections total (default: 100)
- max-conns: max connections per upstream host, 0=unlimited (default: 0)

This allows users to limit concurrent connections to backend servers,
preventing connection exhaustion when upstreams have limited capacity.

Related to #159

Author: umputun

README.md

@@ -333,6 +333,15 @@ Reproxy allows to define system level max req/sec value for the overall system a
 
 User activity limited for both matched and unmatched routes. All unmatched routes considered as a "single destination group" and get a common limiter which is `rate*3`. It means if 10 (req/sec) defined with `--throttle.user=10` the end user will be able to perform up to 30 request pers second for either static assets or unmatched routes. For matched routes this limiter maintained per destination (route), i.e. request proxied to s1.example.com/api will allow 10 r/s and the request proxied to s2.example.com will allow another 10 r/s.
 
+## Upstream connection limits
+
+Reproxy allows configuring upstream connection pool settings to control how many connections are maintained to backend servers:
+
+- `--upstream.max-idle-conns` - Maximum number of idle connections across all upstream hosts. Default: 100.
+- `--upstream.max-conns` - Maximum number of connections per upstream host (0 = unlimited). Default: 0.
+
+Setting `--upstream.max-conns` limits concurrent connections to each backend, which is useful when upstream servers have limited capacity or to prevent connection exhaustion.
+
 ## Basic auth
 
 Reproxy supports basic auth for all requests. This is useful for protecting endpoints during the development and testing, before allowing unrestricted access to them. This functionality is disabled by default and not granular enough to allow for per-route auth. I.e. enabled basic auth will affect all requests.
@@ -505,6 +514,10 @@ throttle:
       --throttle.system=            throttle overall activity' (default: 0) [$THROTTLE_SYSTEM]
       --throttle.user=              limit req/sec per user and per proxy destination (default: 0) [$THROTTLE_USER]
 
+upstream:
+      --upstream.max-idle-conns=    max idle connections total (default: 100) [$UPSTREAM_MAX_IDLE_CONNS]
+      --upstream.max-conns=         max connections per upstream host (0=unlimited) (default: 0) [$UPSTREAM_MAX_CONNS]
+
 plugin:
       --plugin.enabled              enable plugin support [$PLUGIN_ENABLED]
       --plugin.listen=              registration listen on host:port (default: 127.0.0.1:8081) [$PLUGIN_LISTEN]

app/main.go

@@ -146,6 +146,11 @@ var opts struct {
 		User   int `long:"user" env:"USER"  default:"0" description:"limit req/sec per user and per proxy destination"`
 	} `group:"throttle" namespace:"throttle" env-namespace:"THROTTLE"`
 
+	Upstream struct {
+		MaxIdleConns    int `long:"max-idle-conns" env:"MAX_IDLE_CONNS" default:"100" description:"max idle connections total"`
+		MaxConnsPerHost int `long:"max-conns" env:"MAX_CONNS" default:"0" description:"max connections per upstream host (0=unlimited)"`
+	} `group:"upstream" namespace:"upstream" env-namespace:"UPSTREAM"`
+
 	Plugin struct {
 		Enabled bool   `long:"enabled" env:"ENABLED" description:"enable plugin support"`
 		Listen  string `long:"listen" env:"LISTEN" default:"127.0.0.1:8081" description:"registration listen on host:port"`
@@ -291,15 +296,17 @@ func run() error {
 			ExpectContinue: opts.Timeouts.ExpectContinue,
 			ResponseHeader: opts.Timeouts.ResponseHeader,
 		},
-		Metrics:          makeMetrics(ctx, svc),
-		Reporter:         errReporter,
-		PluginConductor:  makePluginConductor(ctx),
-		ThrottleSystem:   opts.Throttle.System * 3,
-		ThrottleUser:     opts.Throttle.User,
-		BasicAuthEnabled: len(basicAuthAllowed) > 0,
-		BasicAuthAllowed: basicAuthAllowed,
-		KeepHost:         opts.KeepHost,
-		OnlyFrom:         makeOnlyFromMiddleware(),
+		Metrics:                 makeMetrics(ctx, svc),
+		Reporter:                errReporter,
+		PluginConductor:         makePluginConductor(ctx),
+		ThrottleSystem:          opts.Throttle.System * 3,
+		ThrottleUser:            opts.Throttle.User,
+		BasicAuthEnabled:        len(basicAuthAllowed) > 0,
+		BasicAuthAllowed:        basicAuthAllowed,
+		KeepHost:                opts.KeepHost,
+		OnlyFrom:                makeOnlyFromMiddleware(),
+		UpstreamMaxIdleConns:    opts.Upstream.MaxIdleConns,
+		UpstreamMaxConnsPerHost: opts.Upstream.MaxConnsPerHost,
 	}
 
 	err = px.Run(ctx)

app/proxy/proxy.go

@@ -58,6 +58,9 @@ type Http struct { // nolint golint
 
 	KeepHost bool
 
+	UpstreamMaxIdleConns    int
+	UpstreamMaxConnsPerHost int
+
 	dnsResolvers []string // used to mock DNS resolvers for testing
 }
 
@@ -247,7 +250,8 @@ func (h *Http) proxyHandler() http.HandlerFunc {
 				KeepAlive: h.Timeouts.KeepAlive,
 			}).DialContext,
 			ForceAttemptHTTP2:     true,
-			MaxIdleConns:          100,
+			MaxIdleConns:          h.UpstreamMaxIdleConns,
+			MaxConnsPerHost:       h.UpstreamMaxConnsPerHost,
 			IdleConnTimeout:       h.Timeouts.IdleConn,
 			TLSHandshakeTimeout:   h.Timeouts.TLSHandshake,
 			ExpectContinueTimeout: h.Timeouts.ExpectContinue,

app/proxy/proxy_test.go

@@ -1029,3 +1029,84 @@ func TestHttp_discoveredServers(t *testing.T) {
 	res := h.discoveredServers(context.Background(), time.Millisecond)
 	assert.Equal(t, []string{"s1", "s2", "s3"}, res)
 }
+
+func TestHttp_UpstreamConfig(t *testing.T) {
+	port := rand.Intn(10000) + 40000
+
+	ds := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprintf(w, "response %s", r.URL.String())
+	}))
+	defer ds.Close()
+
+	svc := discovery.NewService([]discovery.Provider{
+		&provider.Static{Rules: []string{
+			"localhost,^/api/(.*)," + ds.URL + "/test/$1,",
+		}},
+	}, time.Millisecond*10)
+
+	go func() {
+		_ = svc.Run(context.Background())
+	}()
+	time.Sleep(50 * time.Millisecond)
+
+	t.Run("with default upstream values", func(t *testing.T) {
+		h := Http{
+			Timeouts:                Timeouts{ResponseHeader: 200 * time.Millisecond},
+			Address:                 fmt.Sprintf("127.0.0.1:%d", port),
+			AccessLog:               io.Discard,
+			Matcher:                 svc,
+			Metrics:                 mgmt.NewMetrics(),
+			Reporter:                &ErrorReporter{Nice: true},
+			UpstreamMaxIdleConns:    100, // default value
+			UpstreamMaxConnsPerHost: 0,   // unlimited, default
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+		defer cancel()
+
+		go func() {
+			_ = h.Run(ctx)
+		}()
+		time.Sleep(10 * time.Millisecond)
+
+		resp, err := http.Get("http://localhost:" + strconv.Itoa(port) + "/api/something")
+		require.NoError(t, err)
+		defer resp.Body.Close()
+		assert.Equal(t, http.StatusOK, resp.StatusCode)
+
+		body, err := io.ReadAll(resp.Body)
+		require.NoError(t, err)
+		assert.Equal(t, "response /test/something", string(body))
+	})
+
+	t.Run("with custom upstream values", func(t *testing.T) {
+		port2 := rand.Intn(10000) + 40000
+		h := Http{
+			Timeouts:                Timeouts{ResponseHeader: 200 * time.Millisecond},
+			Address:                 fmt.Sprintf("127.0.0.1:%d", port2),
+			AccessLog:               io.Discard,
+			Matcher:                 svc,
+			Metrics:                 mgmt.NewMetrics(),
+			Reporter:                &ErrorReporter{Nice: true},
+			UpstreamMaxIdleConns:    50,
+			UpstreamMaxConnsPerHost: 10,
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+		defer cancel()
+
+		go func() {
+			_ = h.Run(ctx)
+		}()
+		time.Sleep(10 * time.Millisecond)
+
+		resp, err := http.Get("http://localhost:" + strconv.Itoa(port2) + "/api/something")
+		require.NoError(t, err)
+		defer resp.Body.Close()
+		assert.Equal(t, http.StatusOK, resp.StatusCode)
+
+		body, err := io.ReadAll(resp.Body)
+		require.NoError(t, err)
+		assert.Equal(t, "response /test/something", string(body))
+	})
+}