chore: upload embedded dt response difference samples to s3 (#5792)

Author: itsmihir

processor/internal/transformer/destination_transformer/destination_transformer.go

@@ -9,6 +9,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	backendconfig "github.com/rudderlabs/rudder-server/backend-config"
@@ -21,6 +22,7 @@ import (
 	"github.com/samber/lo"
 
 	"github.com/rudderlabs/rudder-go-kit/config"
+	"github.com/rudderlabs/rudder-go-kit/filemanager"
 	"github.com/rudderlabs/rudder-go-kit/logger"
 	"github.com/rudderlabs/rudder-go-kit/stats"
 
@@ -72,15 +74,18 @@ func New(conf *config.Config, log logger.Logger, stat stats.Stats, opts ...Opt)
 	handle.config.maxRetryBackoffInterval = conf.GetReloadableDurationVar(30, time.Second, "Processor.DestinationTransformer.maxRetryBackoffInterval", "Processor.maxRetryBackoffInterval")
 	handle.config.batchSize = conf.GetReloadableIntVar(100, 1, "Processor.DestinationTransformer.batchSize", "Processor.transformBatchSize")
 
-	handle.config.maxLoggedEvents = conf.GetReloadableIntVar(10000, 1, "Processor.DestinationTransformer.maxLoggedEvents")
+	handle.config.maxLoggedEvents = conf.GetReloadableIntVar(100, 1, "Processor.DestinationTransformer.maxLoggedEvents")
 
 	handle.stats.comparisonTime = handle.stat.NewStat("embedded_destination_transform_comparison_time", stats.TimerType)
 	handle.stats.matchedEvents = handle.stat.NewStat("embedded_destination_transform_matched_events", stats.CountType)
 	handle.stats.mismatchedEvents = handle.stat.NewStat("embedded_destination_transform_mismatched_events", stats.CountType)
 
-	handle.loggedEvents = 0
-	handle.loggedEventsMu = sync.Mutex{}
-	handle.loggedFileName = generateLogFileName()
+	var err error
+	handle.samplingFileManager, err = getSamplingUploader(conf, log)
+	if err != nil {
+		log.Errorn("failed to create dt sampling file manager", obskit.Error(err))
+		handle.samplingFileManager = nil
+	}
 
 	handle.config.compactionEnabled = conf.GetReloadableBoolVar(false, "Processor.DestinationTransformer.compactionEnabled", "Transformer.compactionEnabled")
 
@@ -116,9 +121,8 @@ type Client struct {
 		mismatchedEvents stats.Counter
 	}
 
-	loggedEventsMu sync.Mutex
-	loggedEvents   int64
-	loggedFileName string
+	loggedEvents        atomic.Int64
+	samplingFileManager *filemanager.S3Manager
 }
 
 func (d *Client) transform(ctx context.Context, clientEvents []types.TransformerEvent) types.Response {
@@ -391,7 +395,9 @@ func (c *Client) Transform(ctx context.Context, clientEvents []types.Transformer
 		legacyTransformerResponse := c.transform(ctx, clientEvents)
 		embeddedTransformerResponse := impl(ctx, clientEvents)
 
-		c.CompareAndLog(embeddedTransformerResponse, legacyTransformerResponse)
+		go func() {
+			c.CompareAndLog(ctx, embeddedTransformerResponse, legacyTransformerResponse)
+		}()
 
 		return legacyTransformerResponse
 	}
@@ -429,3 +435,32 @@ func (d *Client) getRequestPayload(data []types.TransformerEvent, compactRequest
 	}
 	return jsonrs.Marshal(data)
 }
+
+func getSamplingUploader(conf *config.Config, log logger.Logger) (*filemanager.S3Manager, error) {
+	var (
+		bucket           = conf.GetString("DTSampling.Bucket", "processor-dt-sampling")
+		endpoint         = conf.GetString("DTSampling.Endpoint", "")
+		accessKeyID      = conf.GetStringVar("", "DTSampling.AccessKeyId", "AWS_ACCESS_KEY_ID")
+		accessKey        = conf.GetStringVar("", "DTSampling.AccessKey", "AWS_SECRET_ACCESS_KEY")
+		s3ForcePathStyle = conf.GetBool("DTSampling.S3ForcePathStyle", false)
+		disableSSL       = conf.GetBool("DTSampling.DisableSsl", false)
+		enableSSE        = conf.GetBoolVar(false, "DTSampling.EnableSse", "AWS_ENABLE_SSE")
+		useGlue          = conf.GetBool("DTSampling.UseGlue", false)
+		region           = conf.GetStringVar("us-east-1", "DTSampling.Region", "AWS_DEFAULT_REGION")
+	)
+	s3Config := map[string]any{
+		"bucketName":       bucket,
+		"endpoint":         endpoint,
+		"accessKeyID":      accessKeyID,
+		"accessKey":        accessKey,
+		"s3ForcePathStyle": s3ForcePathStyle,
+		"disableSSL":       disableSSL,
+		"enableSSE":        enableSSE,
+		"useGlue":          useGlue,
+		"region":           region,
+	}
+
+	return filemanager.NewS3Manager(s3Config, log.Withn(logger.NewStringField("component", "dt-uploader")), func() time.Duration {
+		return conf.GetDuration("DTSampling.Timeout", 120, time.Second)
+	})
+}

processor/internal/transformer/destination_transformer/embedded/pubsub/pubsub.go

@@ -110,7 +110,9 @@ func getTopic(event types.TransformerEvent, topicMap map[string]string) (string,
 		return topic, nil
 	}
 
-	return "", fmt.Errorf("no topic set for this event")
+	// Capital "No" needed for mirroring/comparison; re-enable lint after mirroring ends.
+	//nolint:staticcheck
+	return "", fmt.Errorf("No topic set for this event")
 }
 
 func getAttributeKeysFromEvent(event types.TransformerEvent, attributesMap map[string][]string) []string {

processor/internal/transformer/destination_transformer/embedded/pubsub/pubsub_test.go

@@ -370,7 +370,7 @@ func TestTransform(t *testing.T) {
 			want: types.Response{
 				FailedEvents: []types.TransformerResponse{
 					{
-						Error:      "no topic set for this event",
+						Error:      "No topic set for this event",
 						Metadata:   types.Metadata{},
 						StatusCode: http.StatusBadRequest,
 						StatTags:   destinationStatTags,
@@ -416,7 +416,7 @@ func TestTransform(t *testing.T) {
 			want: types.Response{
 				FailedEvents: []types.TransformerResponse{
 					{
-						Error:      "no topic set for this event",
+						Error:      "No topic set for this event",
 						Metadata:   types.Metadata{},
 						StatusCode: http.StatusBadRequest,
 						StatTags:   destinationStatTags,

processor/internal/transformer/destination_transformer/logger.go

@@ -1,48 +1,65 @@
 package destination_transformer
 
 import (
+	"bytes"
+	"context"
 	"fmt"
+	"path"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/uuid"
-	"github.com/samber/lo"
 
+	"github.com/rudderlabs/rudder-go-kit/config"
+	"github.com/rudderlabs/rudder-go-kit/logger"
 	obskit "github.com/rudderlabs/rudder-observability-kit/go/labels"
 
-	"github.com/rudderlabs/rudder-go-kit/stringify"
-
+	"github.com/rudderlabs/rudder-server/jsonrs"
 	"github.com/rudderlabs/rudder-server/processor/types"
-	"github.com/rudderlabs/rudder-server/utils/misc"
 )
 
 func (c *Client) CompareAndLog(
+	ctx context.Context,
 	embeddedResponse, legacyResponse types.Response,
 ) {
-	c.loggedEventsMu.Lock()
-	defer c.loggedEventsMu.Unlock()
+	if c.samplingFileManager == nil { // Cannot upload, we should just report the issue with no diff
+		c.log.Warnn("DestinationTransformer sanity check failed")
+		return
+	}
 
-	if c.loggedEvents >= int64(c.config.maxLoggedEvents.Load()) {
+	if c.loggedEvents.Load() >= int64(c.config.maxLoggedEvents.Load()) {
 		return
 	}
 
-	c.stats.comparisonTime.RecordDuration()()
+	defer c.stats.comparisonTime.RecordDuration()()
 
 	differingResponse, sampleDiff := c.differingEvents(embeddedResponse, legacyResponse)
-	if len(differingResponse) == 0 && sampleDiff == "" {
-		c.log.Infof("Embedded and legacy responses are matches")
+	noOfDifferences := int64(len(differingResponse))
+	if noOfDifferences == 0 && sampleDiff == "" {
+		return
+	}
+
+	c.loggedEvents.Add(noOfDifferences)
+
+	objName := path.Join("embedded-dt-samples", config.GetKubeNamespace(), uuid.New().String())
+	differingResponseJSON, err := jsonrs.Marshal(differingResponse)
+	if err != nil {
+		c.loggedEvents.Add(-noOfDifferences)
+		c.log.Errorn("DestinationTransformer sanity check failed (cannot encode differingResponse)", obskit.Error(err))
 		return
 	}
 
-	logEntries := lo.Map(differingResponse, func(item types.TransformerResponse, index int) string {
-		return stringify.Any(item)
-	})
-	if err := c.write(append([]string{sampleDiff}, logEntries...)); err != nil {
-		c.log.Warnn("Error logging events", obskit.Error(err))
+	// upload sample diff and differing response to s3
+	file, err := c.samplingFileManager.UploadReader(ctx, objName, bytes.NewReader(append([]byte(sampleDiff), differingResponseJSON...)))
+	if err != nil {
+		c.loggedEvents.Add(-noOfDifferences)
+		c.log.Errorn("Error uploading DestinationTransformer sanity check diff file", obskit.Error(err))
 		return
 	}
 
-	c.log.Infof("Successfully logged events: %d", len(logEntries))
-	c.loggedEvents += int64(len(logEntries))
+	c.log.Warnn("DestinationTransformer sanity check failed",
+		logger.NewStringField("location", file.Location),
+		logger.NewStringField("objectName", file.ObjectName),
+	)
 }
 
 func (c *Client) differingEvents(
@@ -95,22 +112,3 @@ func (c *Client) differingEvents(
 	c.stats.mismatchedEvents.Count(differedEventsCount)
 	return differedSampleEvents, sampleDiff
 }
-
-func (c *Client) write(data []string) error {
-	writer, err := misc.CreateGZ(c.loggedFileName)
-	if err != nil {
-		return fmt.Errorf("creating buffered writer: %w", err)
-	}
-	defer func() { _ = writer.Close() }()
-
-	for _, entry := range data {
-		if _, err := writer.Write([]byte(entry + "\n")); err != nil {
-			return fmt.Errorf("writing log entry: %w", err)
-		}
-	}
-	return nil
-}
-
-func generateLogFileName() string {
-	return fmt.Sprintf("destination_transformations_debug_%s.log.gz", uuid.NewString())
-}

processor/processor.go

@@ -2939,7 +2939,7 @@ func (proc *Handle) userTransformAndFilter(
 					}
 
 					if proc.utSamplingFileManager == nil { // Cannot upload, we should just report the issue with no diff
-						log.Errorn("UserTransform sanity check failed")
+						log.Warnn("UserTransform sanity check failed")
 						return
 					}
 
@@ -2969,7 +2969,7 @@ func (proc *Handle) userTransformAndFilter(
 						return
 					}
 
-					log.Errorn("UserTransform sanity check failed",
+					log.Warnn("UserTransform sanity check failed",
 						logger.NewStringField("diffLocation", diffFile.Location),
 						logger.NewStringField("diffObjectName", diffFile.ObjectName),
 						logger.NewStringField("clientEventsLocation", clientEventsFile.Location),