fix: httplb client doesn't respect connection limits (#5773)

# Description

- respecting `MaxConnsPerHost` & `MaxIdleConnsPerHost` in `httplb`
transport
- removed `recycled` client as it can be emulated using `httplb`
- `httplb` client should only use `WithRoundTripperMaxLifetime` in case
`{DEST,USER}_TRANSFORM_URL_IS_HEADLESS=false` or
`Transformer.Client.recycle: true`. If none is present,
`WithRoundTripperMaxLifetime` will not be used

## Security

- [x] The code changed/added as part of this pull request won't create
any security issues with how the software is being used.

Author: atzoum

go.mod

@@ -35,7 +35,7 @@ require (
 	github.com/apache/pulsar-client-go v0.14.0
 	github.com/araddon/dateparse v0.0.0-20210429162001-6b43995a97de
 	github.com/aws/aws-sdk-go v1.55.7
-	github.com/bufbuild/httplb v0.3.1
+	github.com/bufbuild/httplb v0.4.0
 	github.com/cenkalti/backoff v2.2.1+incompatible
 	github.com/cenkalti/backoff/v4 v4.3.0
 	github.com/confluentinc/confluent-kafka-go/v2 v2.8.0

go.sum

@@ -351,8 +351,8 @@ github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdb
 github.com/bsm/gomega v1.26.0/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
 github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
 github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
-github.com/bufbuild/httplb v0.3.1 h1:eY3bDouZyqcyEdUL4/NibxoVh7mXCMKVne1859TDxwQ=
-github.com/bufbuild/httplb v0.3.1/go.mod h1:oMeYRvMM4jbtYhwIwWwKnrnWxe2eeXrp2sqEqaLkuJs=
+github.com/bufbuild/httplb v0.4.0 h1:f3ebi8Dj1tNelLJPf4V5XlblefH/WqwybHYpGODTtbk=
+github.com/bufbuild/httplb v0.4.0/go.mod h1:9XDjl/3UvlkOQUKthLlKn92C1/1SuZ3UCiekxZbenck=
 github.com/buger/goterm v1.0.4 h1:Z9YvGmOih81P0FbVtEYTFF6YsSgxSUKEhf/f9bTMXbY=
 github.com/buger/goterm v1.0.4/go.mod h1:HiFWV3xnkolgrBV3mY8m0X0Pumt4zg4QhbdOzQtB8tE=
 github.com/bytedance/sonic v1.11.6/go.mod h1:LysEHSvpvDySVdC2f87zGWf6CIKJcAvqab1ZaiQtds4=

internal/transformer-client/client.go

@@ -3,7 +3,6 @@
 package transformerclient
 
 import (
-	"context"
 	"net"
 	"net/http"
 	"time"
@@ -12,8 +11,6 @@ import (
 	"github.com/bufbuild/httplb/conn"
 	"github.com/bufbuild/httplb/picker"
 	"github.com/bufbuild/httplb/resolver"
-
-	"github.com/rudderlabs/rudder-server/utils/sysUtils"
 )
 
 type ClientConfig struct {
@@ -26,10 +23,9 @@ type ClientConfig struct {
 
 	ClientTimeout time.Duration //	600*time.Second
 	ClientTTL     time.Duration //	10*time.Second
-
-	ClientType string // stdlib(default), recycled, httplb
-
-	PickerType string // power_of_two(default), round_robin, least_loaded_random, least_loaded_round_robin, random
+	ClientType    string        // stdlib(default), httplb
+	PickerType    string        // power_of_two(default), round_robin, least_loaded_random, least_loaded_round_robin, random
+	Recycle       bool          // false
 }
 
 type Client interface {
@@ -72,22 +68,24 @@ func NewClient(config *ClientConfig) Client {
 	}
 
 	switch config.ClientType {
-	case "stdlib":
-		return client
-	case "recycled":
-		return sysUtils.NewRecycledHTTPClient(func() *http.Client {
-			return client
-		}, clientTTL)
 	case "httplb":
-		return httplb.NewClient(
-			httplb.WithRootContext(context.TODO()),
+		tr := &httplbtransport{
+			MaxConnsPerHost:     transport.MaxConnsPerHost,
+			MaxIdleConnsPerHost: transport.MaxIdleConnsPerHost,
+		}
+		options := []httplb.ClientOption{
 			httplb.WithPicker(getPicker(config.PickerType)),
 			httplb.WithIdleConnectionTimeout(transport.IdleConnTimeout),
 			httplb.WithRequestTimeout(client.Timeout),
-			httplb.WithRoundTripperMaxLifetime(transport.IdleConnTimeout),
-			httplb.WithIdleTransportTimeout(2*transport.IdleConnTimeout),
 			httplb.WithResolver(resolver.NewDNSResolver(net.DefaultResolver, resolver.PreferIPv4, clientTTL)),
-		)
+			httplb.WithTransport("http", tr),
+			httplb.WithTransport("https", tr),
+		}
+		if config.Recycle {
+			options = append(options, httplb.WithRoundTripperMaxLifetime(transport.IdleConnTimeout))
+		}
+
+		return httplb.NewClient(options...)
 	default:
 		return client
 	}
@@ -110,10 +108,27 @@ func getPicker(pickerType string) func(prev picker.Picker, allConns conn.Conns)
 	}
 }
 
-type HTTPLBTransport struct {
+type httplbtransport struct {
+	MaxConnsPerHost     int
+	MaxIdleConnsPerHost int
 	*http.Transport
 }
 
-func (t *HTTPLBTransport) NewRoundTripper(scheme, target string, config httplb.TransportConfig) httplb.RoundTripperResult {
-	return httplb.RoundTripperResult{RoundTripper: t.Transport, Close: t.CloseIdleConnections}
+func (s httplbtransport) NewRoundTripper(_, _ string, opts httplb.TransportConfig) httplb.RoundTripperResult {
+	transport := &http.Transport{
+		Proxy:                  opts.ProxyFunc,
+		GetProxyConnectHeader:  opts.ProxyConnectHeadersFunc,
+		DialContext:            opts.DialFunc,
+		ForceAttemptHTTP2:      true,
+		MaxConnsPerHost:        s.MaxConnsPerHost,
+		MaxIdleConns:           s.MaxIdleConnsPerHost,
+		MaxIdleConnsPerHost:    s.MaxIdleConnsPerHost,
+		IdleConnTimeout:        opts.IdleConnTimeout,
+		TLSHandshakeTimeout:    opts.TLSHandshakeTimeout,
+		TLSClientConfig:        opts.TLSClientConfig,
+		MaxResponseHeaderBytes: opts.MaxResponseHeaderBytes,
+		ExpectContinueTimeout:  1 * time.Second,
+		DisableCompression:     opts.DisableCompression,
+	}
+	return httplb.RoundTripperResult{RoundTripper: transport, Close: transport.CloseIdleConnections}
 }

processor/internal/transformer/destination_transformer/destination_transformer.go

@@ -64,7 +64,7 @@ func New(conf *config.Config, log logger.Logger, stat stats.Stats, opts ...Opt)
 	handle.conf = conf
 	handle.log = log
 	handle.stat = stat
-	handle.client = transformerclient.NewClient(transformerutils.TransformerClientConfig(conf, "DestinationTransformer"))
+	handle.client = transformerclient.NewClient(transformerutils.TransformerClientConfig(conf, "DestinationTransformer", conf.GetBoolVar(true, "DEST_TRANSFORM_URL_IS_HEADLESS")))
 	handle.config.destTransformationURL = handle.conf.GetString("DEST_TRANSFORM_URL", "http://localhost:9090")
 	handle.config.timeoutDuration = conf.GetDuration("HttpClient.procTransformer.timeout", 600, time.Second)
 	handle.config.maxRetry = conf.GetReloadableIntVar(30, 1, "Processor.DestinationTransformer.maxRetry", "Processor.maxRetry")

processor/internal/transformer/trackingplan_validation/trackingplan_validation.go

@@ -42,7 +42,7 @@ func New(conf *config.Config, log logger.Logger, stat stats.Stats, opts ...Opt)
 	handle.conf = conf
 	handle.log = log
 	handle.stat = stat
-	handle.client = transformerclient.NewClient(transformerutils.TransformerClientConfig(conf, "TrackingPlanValidation"))
+	handle.client = transformerclient.NewClient(transformerutils.TransformerClientConfig(conf, "TrackingPlanValidation", conf.GetBoolVar(true, "DEST_TRANSFORM_URL_IS_HEADLESS")))
 	handle.config.destTransformationURL = handle.conf.GetString("DEST_TRANSFORM_URL", "http://localhost:9090")
 	handle.config.maxRetry = conf.GetReloadableIntVar(30, 1, "Processor.TrackingPlanValidation.maxRetry", "Processor.maxRetry")
 	handle.config.timeoutDuration = conf.GetDuration("HttpClient.procTransformer.timeout", 600, time.Second)

processor/internal/transformer/user_transformer/user_transformer.go

@@ -43,7 +43,7 @@ func New(conf *config.Config, log logger.Logger, stat stats.Stats, opts ...Opt)
 	handle.conf = conf
 	handle.log = log.Child("user_transformer")
 	handle.stat = stat
-	handle.client = transformerclient.NewClient(transformerutils.TransformerClientConfig(conf, "UserTransformer"))
+	handle.client = transformerclient.NewClient(transformerutils.TransformerClientConfig(conf, "UserTransformer", conf.GetBoolVar(true, "USER_TRANSFORM_URL_IS_HEADLESS")))
 	handle.config.userTransformationURL = handle.conf.GetString("USER_TRANSFORM_URL", handle.conf.GetString("DEST_TRANSFORM_URL", "http://localhost:9090"))
 	handle.config.timeoutDuration = conf.GetDuration("HttpClient.procTransformer.timeout", 600, time.Second)
 	handle.config.failOnUserTransformTimeout = conf.GetReloadableBoolVar(false, "Processor.UserTransformer.failOnUserTransformTimeout", "Processor.Transformer.failOnUserTransformTimeout")

processor/internal/transformer/utils.go

@@ -26,7 +26,7 @@ func IsJobTerminated(status int) bool {
 	return status >= http.StatusOK && status < http.StatusInternalServerError
 }
 
-func TransformerClientConfig(conf *config.Config, configPrefix string) *transformerclient.ClientConfig {
+func TransformerClientConfig(conf *config.Config, configPrefix string, headless bool) *transformerclient.ClientConfig {
 	transformerClientConfig := &transformerclient.ClientConfig{
 		ClientTimeout: conf.GetDurationVar(600, time.Second, fmt.Sprintf("HttpClient.procTransformer.%s.timeout", configPrefix), "HttpClient.procTransformer.timeout"),
 		ClientTTL:     conf.GetDurationVar(10, time.Second, fmt.Sprintf("Transformer.Client.%s.ttl", configPrefix), "Transformer.Client.ttl"),
@@ -37,6 +37,7 @@ func TransformerClientConfig(conf *config.Config, configPrefix string) *transfor
 	transformerClientConfig.TransportConfig.MaxConnsPerHost = conf.GetIntVar(100, 1, fmt.Sprintf("Transformer.Client.%s.maxHTTPConnections", configPrefix), "Transformer.Client.maxHTTPConnections")
 	transformerClientConfig.TransportConfig.MaxIdleConnsPerHost = conf.GetIntVar(10, 1, fmt.Sprintf("Transformer.Client.%s.maxHTTPIdleConnections", configPrefix), "Transformer.Client.maxHTTPIdleConnections")
 	transformerClientConfig.TransportConfig.IdleConnTimeout = conf.GetDurationVar(30, time.Second, fmt.Sprintf("Transformer.Client.%s.maxIdleConnDuration", configPrefix), "Transformer.Client.maxIdleConnDuration")
+	transformerClientConfig.Recycle = !headless || conf.GetBoolVar(false, fmt.Sprintf("Transformer.Client.%s.recycle", configPrefix), "Transformer.Client.recycle")
 	return transformerClientConfig
 }
 

router/transformer/transformer.go

@@ -620,6 +620,7 @@ func (trans *handle) transformerClientConfig() *transformerclient.ClientConfig {
 	transformerClientConfig.TransportConfig.MaxConnsPerHost = config.GetIntVar(100, 1, "Transformer.Client.maxHTTPConnections")
 	transformerClientConfig.TransportConfig.MaxIdleConnsPerHost = config.GetIntVar(10, 1, "Transformer.Client.maxHTTPIdleConnections")
 	transformerClientConfig.TransportConfig.IdleConnTimeout = config.GetDurationVar(30, time.Second, "Transformer.Client.maxIdleConnDuration")
+	transformerClientConfig.Recycle = !config.GetBoolVar(true, "DEST_TRANSFORM_URL_IS_HEADLESS") || config.GetBoolVar(false, "Transformer.Client.DestinationTransformer.recycle", "Transformer.Client.recycle")
 	return transformerClientConfig
 }
 

utils/sysUtils/httpclient.go

@@ -3,44 +3,9 @@ package sysUtils
 
 import (
 	"net/http"
-	"sync"
-	"time"
 )
 
 // HTTPClient interface
 type HTTPClientI interface {
 	Do(req *http.Request) (*http.Response, error)
 }
-
-type RecycledHTTPClient struct {
-	client          *http.Client
-	lastRefreshTime time.Time
-	ttl             time.Duration
-	clientFunc      func() *http.Client
-	lock            sync.Mutex
-}
-
-func NewRecycledHTTPClient(_clientFunc func() *http.Client, _ttl time.Duration) *RecycledHTTPClient {
-	return &RecycledHTTPClient{
-		client:          _clientFunc(),
-		clientFunc:      _clientFunc,
-		ttl:             _ttl,
-		lastRefreshTime: time.Now(),
-	}
-}
-
-func (r *RecycledHTTPClient) GetClient() *http.Client {
-	r.lock.Lock()
-	defer r.lock.Unlock()
-
-	if r.ttl > 0 && time.Since(r.lastRefreshTime) > r.ttl {
-		r.client.CloseIdleConnections()
-		r.client = r.clientFunc()
-		r.lastRefreshTime = time.Now()
-	}
-	return r.client
-}
-
-func (r *RecycledHTTPClient) Do(req *http.Request) (*http.Response, error) {
-	return r.GetClient().Do(req)
-}