fix: refresh connections to avoid hotspotting of load distribution

processor/transformer/transformer.go

@@ -26,6 +26,7 @@ import (
 	backendconfig "github.com/rudderlabs/rudder-server/backend-config"
 	"github.com/rudderlabs/rudder-server/processor/integrations"
 	"github.com/rudderlabs/rudder-server/utils/httputil"
+	"github.com/rudderlabs/rudder-server/utils/sysUtils"
 	"github.com/rudderlabs/rudder-server/utils/types"
 	warehouseutils "github.com/rudderlabs/rudder-server/warehouse/utils"
 )
@@ -136,7 +137,7 @@ type Opt func(*handle)
 
 func WithClient(client *http.Client) Opt {
 	return func(s *handle) {
-		s.client = client
+		s.recycledClient = sysUtils.NewRecycledHTTPClient(func() *http.Client { return client }, 0)
 	}
 }
 
@@ -157,7 +158,7 @@ type handle struct {
 	logger logger.Logger
 	stat   stats.Stats
 
-	client *http.Client
+	recycledClient *sysUtils.RecycledHTTPClient
 
 	guardConcurrency chan struct{}
 
@@ -209,16 +210,18 @@ func NewTransformer(conf *config.Config, log logger.Logger, stat stats.Stats, op
 
 	trans.guardConcurrency = make(chan struct{}, trans.config.maxConcurrency)
 
-	if trans.client == nil {
-		trans.client = &http.Client{
-			Transport: &http.Transport{
-				DisableKeepAlives:   trans.config.disableKeepAlives,
-				MaxConnsPerHost:     trans.config.maxHTTPConnections,
-				MaxIdleConnsPerHost: trans.config.maxHTTPIdleConnections,
-				IdleConnTimeout:     trans.config.maxIdleConnDuration,
-			},
-			Timeout: trans.config.timeoutDuration,
-		}
+	if trans.recycledClient == nil {
+		trans.recycledClient = sysUtils.NewRecycledHTTPClient(func() *http.Client {
+			return &http.Client{
+				Transport: &http.Transport{
+					DisableKeepAlives:   trans.config.disableKeepAlives,
+					MaxConnsPerHost:     trans.config.maxHTTPConnections,
+					MaxIdleConnsPerHost: trans.config.maxHTTPIdleConnections,
+					IdleConnTimeout:     trans.config.maxIdleConnDuration,
+				},
+				Timeout: trans.config.timeoutDuration,
+			}
+		}, config.GetDuration("Transformer.Client.ttl", 120, time.Second))
 	}
 
 	for _, opt := range opts {
@@ -472,7 +475,7 @@ func (trans *handle) doPost(ctx context.Context, rawJSON []byte, url, stage stri
 				// Header to let transformer know that the client understands event filter code
 				req.Header.Set("X-Feature-Filter-Code", "?1")
 
-				resp, reqErr = trans.client.Do(req)
+				resp, reqErr = trans.recycledClient.GetClient().Do(req)
 			})
 			trans.stat.NewTaggedStat("processor.transformer_request_time", stats.TimerType, tags).SendTiming(time.Since(requestStartTime))
 			if reqErr != nil {

processor/transformer/transformer_test.go

@@ -19,6 +19,7 @@ import (
 	"github.com/rudderlabs/rudder-server/testhelper/backendconfigtest"
 	warehouseutils "github.com/rudderlabs/rudder-server/warehouse/utils"
 
+	"github.com/rudderlabs/rudder-server/utils/sysUtils"
 	"github.com/rudderlabs/rudder-server/utils/types"
 
 	"github.com/stretchr/testify/require"
@@ -173,7 +174,7 @@ func TestTransformer(t *testing.T) {
 			tr.stat = statsStore
 			tr.logger = logger.NOP
 			tr.conf = config.Default
-			tr.client = srv.Client()
+			tr.recycledClient = sysUtils.NewRecycledHTTPClient(srv.Client, 0)
 			tr.guardConcurrency = make(chan struct{}, 200)
 			tr.sentStat = tr.stat.NewStat("transformer_sent", stats.CountType)
 			tr.receivedStat = tr.stat.NewStat("transformer_received", stats.CountType)
@@ -359,7 +360,7 @@ func TestTransformer(t *testing.T) {
 				tr.stat = stats.Default
 				tr.logger = logger.NOP
 				tr.conf = config.Default
-				tr.client = client
+				tr.recycledClient = sysUtils.NewRecycledHTTPClient(func() *http.Client { return client }, 0)
 				tr.config.maxRetry = config.SingleValueLoader(tc.retries)
 				tr.config.failOnUserTransformTimeout = config.SingleValueLoader(tc.failOnUserTransformTimeout)
 				tr.cpDownGauge = tr.stat.NewStat("control_plane_down", stats.GaugeType)
@@ -422,7 +423,7 @@ func TestTransformer(t *testing.T) {
 		tr.stat = stats.Default
 		tr.logger = logger.NOP
 		tr.conf = config.Default
-		tr.client = srv.Client()
+		tr.recycledClient = sysUtils.NewRecycledHTTPClient(srv.Client, 0)
 		tr.config.maxRetry = config.SingleValueLoader(1)
 		tr.config.maxRetryBackoffInterval = config.SingleValueLoader(1 * time.Second)
 		tr.config.timeoutDuration = 1 * time.Second
@@ -553,7 +554,7 @@ func TestTransformer(t *testing.T) {
 				tr.stat = stats.Default
 				tr.logger = logger.NOP
 				tr.conf = config.Default
-				tr.client = srv.Client()
+				tr.recycledClient = sysUtils.NewRecycledHTTPClient(srv.Client, 0)
 				tr.config.failOnUserTransformTimeout = config.SingleValueLoader(false)
 				tr.config.maxRetry = config.SingleValueLoader(tc.retries)
 				tr.config.failOnError = config.SingleValueLoader(tc.failOnError)
@@ -653,7 +654,7 @@ func TestTransformer(t *testing.T) {
 				defer srv.Close()
 
 				tr := handle{}
-				tr.client = srv.Client()
+				tr.recycledClient = sysUtils.NewRecycledHTTPClient(srv.Client, 0)
 				tr.stat = stats.Default
 				tr.conf = config.Default
 				tr.logger = logger.NOP

router/transformer/transformer.go

@@ -49,8 +49,8 @@ const (
 // handle is the handle for this class
 type handle struct {
 	tr *http.Transport
-	// http client for router transformation request
-	client *http.Client
+	// http recycledClient for router transformation request
+	recycledClient *sysUtils.RecycledHTTPClient
 	// Mockable http.client for transformer proxy request
 	proxyClient sysUtils.HTTPClientI
 	// http client timeout for transformer proxy request
@@ -179,7 +179,7 @@ func (trans *handle) Transform(transformType string, transformMessage *types.Tra
 			req = req.WithContext(cntx.CtxWithDestInfo(req.Context(), destinationInfo))
 			resp, err = trans.clientOAuthV2.Do(req)
 		} else {
-			resp, err = trans.client.Do(req)
+			resp, err = trans.recycledClient.GetClient().Do(req)
 		}
 
 		if err == nil {
@@ -500,7 +500,10 @@ func (trans *handle) setup(destinationTimeout, transformTimeout time.Duration, c
 	// Basically this timeout we will configure when we make final call to destination to send event
 	trans.destinationTimeout = destinationTimeout
 	// This client is used for Router Transformation
-	trans.client = &http.Client{Transport: trans.tr, Timeout: trans.transformTimeout}
+	trans.recycledClient = sysUtils.NewRecycledHTTPClient(
+		func() *http.Client {
+			return &http.Client{Transport: trans.tr.Clone(), Timeout: trans.transformTimeout}
+		}, config.GetDuration("Transformer.Client.ttl", 120, time.Second))
 	optionalArgs := &oauthv2httpclient.HttpClientOptionalArgs{
 		Locker:             locker,
 		Augmenter:          extensions.RouterBodyAugmenter,

utils/sysUtils/httpclient.go

@@ -3,9 +3,40 @@ 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
+}