feat: introduce httplb for ut

go.mod

@@ -35,6 +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.5
+	github.com/bufbuild/httplb v0.3.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.6.1

go.sum

@@ -357,6 +357,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.0 h1:sCMPD+89ydD3atcVareDsiv/kUT+pLHolENMoCGZJV8=
+github.com/bufbuild/httplb v0.3.0/go.mod h1:qDNs7dSFxIhKi/DA/rCCPVzbQfHs1JVxPMl9EvrbL4Q=
 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=

processor/transformer/embeded/pubsub.go

@@ -0,0 +1,87 @@
+package embeded
+
+import (
+	"context"
+	"strings"
+
+	"github.com/rudderlabs/rudder-server/processor/transformer"
+)
+
+type PubSubTransformer struct{}
+
+func (t *PubSubTransformer) Transform(ctx context.Context, clientEvents []transformer.TransformerEvent, batchSize int) transformer.Response {
+	var response transformer.Response
+
+	for _, clientEvent := range clientEvents {
+		msg := clientEvent.Message
+
+		// Get topic ID for the event
+		topicID := getTopicID(clientEvent)
+		if topicID == "" {
+			response.FailedEvents = append(response.FailedEvents, transformer.TransformerResponse{
+				Output: clientEvent.Message,
+			})
+			continue
+		}
+
+		// Create attributes metadata
+		attributes := createAttributesMetadata(clientEvent)
+
+		// Create transformed event
+		transformedEvent := transformer.TransformerResponse{
+			Output: map[string]interface{}{
+				"message":    msg,
+				"topic_id":   topicID,
+				"attributes": attributes,
+				"user_id":    getEventUserID(clientEvent),
+			},
+		}
+
+		response.Events = append(response.Events, transformedEvent)
+	}
+
+	return response
+}
+
+func getTopicID(event transformer.TransformerEvent) string {
+	// Get topic mapping from config
+	topicMap := event.Destination.Config["eventToTopicMap"].(map[string]string)
+
+	if event.Message["type"] == "" {
+		return ""
+	}
+
+	// Try to match event name first, then type, then fallback to wildcard
+	if event.Message["event"] != "" {
+		if topic, ok := topicMap[strings.ToLower(event.Message["event"].(string))]; ok {
+			return topic
+		}
+	}
+
+	if topic, ok := topicMap[strings.ToLower(event.Message["type"].(string))]; ok {
+		return topic
+	}
+
+	return topicMap["*"]
+}
+
+func createAttributesMetadata(event transformer.TransformerEvent) map[string]string {
+	attributes := make(map[string]string)
+
+	// Get attributes mapping from config
+	attrMap := event.Destination.Config["eventToAttributesMap"].(map[string]string)
+	if len(attrMap) == 0 {
+		return attributes
+	}
+
+	// ... rest of attributes logic ...
+
+	return attributes
+}
+
+func getEventUserID(event transformer.TransformerEvent) string {
+	if event.Message["user_id"] != "" {
+		return event.Message["user_id"].(string)
+	}
+	return event.Message["anonymous_id"].(string)
+}

processor/transformer/transformer.go

@@ -7,6 +7,7 @@
 	"context"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
 	"os"
 	"runtime/trace"
@@ -15,6 +16,8 @@
 	"sync"
 	"time"
 
+	"github.com/bufbuild/httplb"
+	"github.com/bufbuild/httplb/resolver"
 	"github.com/cenkalti/backoff"
 	jsoniter "github.com/json-iterator/go"
 	"github.com/samber/lo"
@@ -26,6 +29,7 @@
 	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,9 +140,9 @@
 
 type Opt func(*handle)
 
-func WithClient(client *http.Client) Opt {
+func WithClient(client HTTPDoer) Opt {
 	return func(s *handle) {
-		s.client = client
+		s.httpClient = client
 	}
 }
 
@@ -149,6 +153,10 @@
 	Validate(ctx context.Context, clientEvents []TransformerEvent, batchSize int) Response
 }
 
+type HTTPDoer interface {
+	Do(req *http.Request) (*http.Response, error)
+}
+
 // handle is the handle for this class
 type handle struct {
 	sentStat     stats.Measurement
@@ -159,7 +167,7 @@
 	logger logger.Logger
 	stat   stats.Stats
 
-	client *http.Client
+	httpClient HTTPDoer
 
 	guardConcurrency chan struct{}
 
@@ -211,8 +219,11 @@
 
 	trans.guardConcurrency = make(chan struct{}, trans.config.maxConcurrency)
 
-	if trans.client == nil {
-		trans.client = &http.Client{
+	clientType := config.GetString("Transformer.Client.type", "stdlib")
+
+	switch clientType {
+	case "stdlib":
+		trans.httpClient = &http.Client{
 			Transport: &http.Transport{
 				DisableKeepAlives:   trans.config.disableKeepAlives,
 				MaxConnsPerHost:     trans.config.maxHTTPConnections,
@@ -221,6 +232,38 @@
 			},
 			Timeout: trans.config.timeoutDuration,
 		}
+	case "recycled":
+		trans.httpClient = 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))
+	case "httplb":
+		trans.httpClient = httplb.NewClient(
+			httplb.WithTransport("http", &HTTPLBTransport{
+				Transport: &http.Transport{
+					DisableKeepAlives:   trans.config.disableKeepAlives,
+					MaxConnsPerHost:     trans.config.maxHTTPConnections,
+					MaxIdleConnsPerHost: trans.config.maxHTTPIdleConnections,
+					IdleConnTimeout:     trans.config.maxIdleConnDuration,
+				},
+			}),
+			httplb.WithResolver(
+				resolver.NewDNSResolver(
+					net.DefaultResolver,
+					resolver.PreferIPv6,
+					config.GetDuration("Transformer.Client.ttl", 120, time.Second), // TTL value
+				),
+			),
+		)
+	default:
+		panic(fmt.Sprintf("unknown transformer client type: %s", clientType))
 	}
 
 	for _, opt := range opts {
@@ -245,6 +288,14 @@
 	return trans.transform(ctx, clientEvents, trans.trackingPlanValidationURL(), batchSize, trackingPlanValidationStage)
 }
 
+type HTTPLBTransport struct {
+	*http.Transport
+}
+
+func (t *HTTPLBTransport) NewRoundTripper(scheme, target string, config httplb.TransportConfig) httplb.RoundTripperResult {
+	return httplb.RoundTripperResult{RoundTripper: t.Transport, Close: t.CloseIdleConnections}
+}
+
 func (trans *handle) transform(
 	ctx context.Context,
 	clientEvents []TransformerEvent,
@@ -474,7 +525,7 @@
 				// 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.httpClient.Do(req)
 			})
 			trans.stat.NewTaggedStat("processor.transformer_request_time", stats.TimerType, tags).SendTiming(time.Since(requestStartTime))
 			if reqErr != nil {
@@ -495,7 +546,6 @@
 			retryCount++
 			trans.logger.Warnn(
 				"JS HTTP connection error",
-				logger.NewStringField("URL", url),
 				logger.NewErrorField(err),
 				logger.NewIntField("attempts", int64(retryCount)),
 			)

processor/transformer/transformer_test.go

@@ -173,7 +173,7 @@ func TestTransformer(t *testing.T) {
 			tr.stat = statsStore
 			tr.logger = logger.NOP
 			tr.conf = config.Default
-			tr.client = srv.Client()
+			tr.httpClient = srv.Client()
 			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 +359,7 @@ func TestTransformer(t *testing.T) {
 				tr.stat = stats.Default
 				tr.logger = logger.NOP
 				tr.conf = config.Default
-				tr.client = client
+				tr.httpClient = client
 				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 +422,7 @@ func TestTransformer(t *testing.T) {
 		tr.stat = stats.Default
 		tr.logger = logger.NOP
 		tr.conf = config.Default
-		tr.client = srv.Client()
+		tr.httpClient = srv.Client()
 		tr.config.maxRetry = config.SingleValueLoader(1)
 		tr.config.maxRetryBackoffInterval = config.SingleValueLoader(1 * time.Second)
 		tr.config.timeoutDuration = 1 * time.Second
@@ -553,7 +553,7 @@ func TestTransformer(t *testing.T) {
 				tr.stat = stats.Default
 				tr.logger = logger.NOP
 				tr.conf = config.Default
-				tr.client = srv.Client()
+				tr.httpClient = srv.Client()
 				tr.config.failOnUserTransformTimeout = config.SingleValueLoader(false)
 				tr.config.maxRetry = config.SingleValueLoader(tc.retries)
 				tr.config.failOnError = config.SingleValueLoader(tc.failOnError)
@@ -653,7 +653,7 @@ func TestTransformer(t *testing.T) {
 				defer srv.Close()
 
 				tr := handle{}
-				tr.client = srv.Client()
+				tr.httpClient = srv.Client()
 				tr.stat = stats.Default
 				tr.conf = config.Default
 				tr.logger = logger.NOP

utils/sysUtils/httpclient.go

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