chore: add mocks client

rudderlabs · Feb 12, 2025 · 526c759 · 526c759
1 parent 1ff1da4
commit 526c759
Show file tree

Hide file tree

Showing 3 changed files with 75 additions and 0 deletions.
diff --git a/internal/transformer-client/client.go b/internal/transformer-client/client.go
@@ -1,3 +1,5 @@
+//go:generate mockgen -destination=../../mocks/transformer-client/mock_transformer_client.go -package=mocks_transformer_client github.com/rudderlabs/rudder-server/internal/transformer-client Client
+
 package transformerclient
 
 import (

diff --git a/mocks/transformer-client/mock_transformer_client.go b/mocks/transformer-client/mock_transformer_client.go
diff --git a/processor/internal/user_transformer/user_transformer.go b/processor/internal/user_transformer/user_transformer.go
@@ -8,6 +8,7 @@ import (
 	"net/http"
 	"os"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -37,6 +38,7 @@ type UserTransformer struct {
 		failOnError                config.ValueLoader[bool]
 		maxRetryBackoffInterval    config.ValueLoader[time.Duration]
 		timeoutDuration            time.Duration
+		collectInstanceLevelStats  bool
 		maxConcurrency             int
 	}
 	conf             *config.Config
@@ -69,6 +71,7 @@ func NewUserTransformer(conf *config.Config, log logger.Logger, stat stats.Stats
 	handle.config.maxRetry = conf.GetReloadableIntVar(30, 1, "Processor.maxRetry")
 	handle.config.failOnError = conf.GetReloadableBoolVar(false, "Processor.Transformer.failOnError")
 	handle.config.maxRetryBackoffInterval = conf.GetReloadableDurationVar(30, time.Second, "Processor.maxRetryBackoffInterval")
+	handle.config.collectInstanceLevelStats = conf.GetBool("Processor.collectInstanceLevelStats", false)
 	return handle
 }
 
@@ -291,10 +294,24 @@ func (u *UserTransformer) doPost(ctx context.Context, rawJSON []byte, url string
 
 			resp, reqErr = u.client.Do(req)
 			defer func() { httputil.CloseResponse(resp) }()
+			duration := time.Since(requestStartTime)
 			u.stat.NewTaggedStat("processor.transformer_request_time", stats.TimerType, tags).SendTiming(time.Since(requestStartTime))
 			if reqErr != nil {
 				return reqErr
 			}
+			headerResponseTime := resp.Header.Get("X-Response-Time")
+			instanceWorker := resp.Header.Get("X-Instance-ID")
+
+			if u.config.collectInstanceLevelStats && instanceWorker != "" {
+				newTags := lo.Assign(tags)
+				newTags["instanceWorker"] = instanceWorker
+				dur := duration.Milliseconds()
+				headerTime, err := strconv.ParseFloat(strings.TrimSuffix(headerResponseTime, "ms"), 64)
+				if err == nil {
+					diff := float64(dur) - headerTime
+					u.stat.NewTaggedStat("processor_transform_duration_diff_time", stats.TimerType, newTags).SendTiming(time.Duration(diff) * time.Millisecond)
+				}
+			}
 
 			if !transformer_utils.IsJobTerminated(resp.StatusCode) && resp.StatusCode != transformer_utils.StatusCPDown {
 				return fmt.Errorf("transformer returned status code: %v", resp.StatusCode)