fix: warehouse transformations SourceDefinitionType as empty (#5510)

Author: achettyiitr

go.mod

@@ -51,6 +51,7 @@ require (
 	github.com/golang-migrate/migrate/v4 v4.18.2
 	github.com/golang/mock v1.6.0
 	github.com/gomodule/redigo v1.9.2
+	github.com/google/go-cmp v0.6.0
 	github.com/google/uuid v1.6.0
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.24.0
 	github.com/hashicorp/go-retryablehttp v0.7.7
@@ -231,7 +232,6 @@ require (
 	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/google/flatbuffers v24.12.23+incompatible // indirect
-	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/google/pprof v0.0.0-20241210010833-40e02aabc2ad // indirect
 	github.com/google/s2a-go v0.1.8 // indirect

runner/buckets.go

@@ -146,6 +146,9 @@ var (
 			0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1,
 		},
 
+		"warehouse_dest_transform_matched_events": {
+			0.1, 0.25, 0.5, 1, 2.5, 5, 7.5, 10, 12.5, 15, 30, 60, 120, 300, 600, 900, 1800, 3600,
+		},
 		"warehouse_dest_transform_mismatched_events": {
 			0.1, 0.25, 0.5, 1, 2.5, 5, 7.5, 10, 12.5, 15, 30, 60, 120, 300, 600, 900, 1800, 3600,
 		},

warehouse/transformer/logger.go

@@ -2,7 +2,6 @@ package transformer
 
 import (
 	"fmt"
-	"reflect"
 
 	"github.com/google/uuid"
 	"github.com/samber/lo"
@@ -15,6 +14,8 @@ import (
 	ptrans "github.com/rudderlabs/rudder-server/processor/transformer"
 	"github.com/rudderlabs/rudder-server/utils/misc"
 	"github.com/rudderlabs/rudder-server/utils/types"
+
+	"github.com/google/go-cmp/cmp"
 )
 
 func (t *Transformer) CompareAndLog(
@@ -36,7 +37,7 @@ func (t *Transformer) CompareAndLog(
 
 	t.stats.comparisionTime.RecordDuration()()
 
-	differingEvents := t.differingEvents(events, pResponse, wResponse, eventsByMessageID)
+	differingEvents, sampleDiff := t.differingEvents(events, pResponse, wResponse, eventsByMessageID)
 	if len(differingEvents) == 0 {
 		return
 	}
@@ -47,7 +48,7 @@ func (t *Transformer) CompareAndLog(
 			Metadata: *metadata,
 		})
 	})
-	if err := t.writeLogEntries(logEntries); err != nil {
+	if err := t.write(append([]string{sampleDiff}, logEntries...)); err != nil {
 		t.logger.Warnn("Error logging events", obskit.Error(err))
 		return
 	}
@@ -60,45 +61,50 @@ func (t *Transformer) differingEvents(
 	eventsToTransform []ptrans.TransformerEvent,
 	pResponse, wResponse ptrans.Response,
 	eventsByMessageID map[string]types.SingularEventWithReceivedAt,
-) []types.SingularEventT {
+) ([]types.SingularEventT, string) {
 	// If the event counts differ, return all events in the transformation
 	if len(pResponse.Events) != len(wResponse.Events) || len(pResponse.FailedEvents) != len(wResponse.FailedEvents) {
 		events := lo.Map(eventsToTransform, func(e ptrans.TransformerEvent, _ int) types.SingularEventT {
 			return eventsByMessageID[e.Metadata.MessageID].SingularEvent
 		})
 		t.stats.mismatchedEvents.Observe(float64(len(events)))
-		return events
+		return events, ""
 	}
 
 	var (
 		differedSampleEvents []types.SingularEventT
 		differedEventsCount  int
+		sampleDiff           string
 	)
 
 	for i := range pResponse.Events {
-		if reflect.DeepEqual(pResponse.Events[i], wResponse.Events[i]) {
+		diff := cmp.Diff(wResponse.Events[i], pResponse.Events[i])
+		if len(diff) == 0 {
 			continue
 		}
+
 		if differedEventsCount == 0 {
 			// Collect the mismatched messages and break (sample only)
 			differedSampleEvents = append(differedSampleEvents, lo.Map(pResponse.Events[i].Metadata.GetMessagesIDs(), func(msgID string, _ int) types.SingularEventT {
 				return eventsByMessageID[msgID].SingularEvent
 			})...)
+			sampleDiff = diff
 		}
 		differedEventsCount++
 	}
+	t.stats.matchedEvents.Observe(float64(len(pResponse.Events) - differedEventsCount))
 	t.stats.mismatchedEvents.Observe(float64(differedEventsCount))
-	return differedSampleEvents
+	return differedSampleEvents, sampleDiff
 }
 
-func (t *Transformer) writeLogEntries(entries []string) error {
+func (t *Transformer) write(data []string) error {
 	writer, err := misc.CreateGZ(t.loggedFileName)
 	if err != nil {
 		return fmt.Errorf("creating buffered writer: %w", err)
 	}
 	defer func() { _ = writer.Close() }()
 
-	for _, entry := range entries {
+	for _, entry := range data {
 		if _, err := writer.Write([]byte(entry + "\n")); err != nil {
 			return fmt.Errorf("writing log entry: %w", err)
 		}
@@ -107,5 +113,5 @@ func (t *Transformer) writeLogEntries(entries []string) error {
 }
 
 func generateLogFileName() string {
-	return fmt.Sprintf("warehouse_transformations_debug_%s.log", uuid.NewString())
+	return fmt.Sprintf("warehouse_transformations_debug_%s.log.gz", uuid.NewString())
 }

warehouse/transformer/transformer.go

@@ -37,6 +37,7 @@ func New(conf *config.Config, logger logger.Logger, statsFactory stats.Stats) *T
 		loggedFileName: generateLogFileName(),
 	}
 
+	t.stats.matchedEvents = t.statsFactory.NewStat("warehouse_dest_transform_matched_events", stats.HistogramType)
 	t.stats.mismatchedEvents = t.statsFactory.NewStat("warehouse_dest_transform_mismatched_events", stats.HistogramType)
 	t.stats.comparisionTime = t.statsFactory.NewStat("warehouse_dest_transform_comparison_time", stats.TimerType)
 
@@ -80,6 +81,7 @@ func (t *Transformer) Transform(_ context.Context, clientEvents []ptrans.Transfo
 		}
 
 		res.Events = append(res.Events, lo.Map(r, func(item map[string]any, index int) ptrans.TransformerResponse {
+			clientEvent.Metadata.SourceDefinitionType = "" // TODO: Currently, it's getting ignored during JSON marshalling Remove this once we start using it.
 			return ptrans.TransformerResponse{
 				Output:     item,
 				Metadata:   clientEvent.Metadata,

warehouse/transformer/transformer_test.go

@@ -626,7 +626,7 @@ func getTrackMetadata(destinationType, sourceCategory string) ptrans.Metadata {
 }
 
 func TestTransformer_CompareAndLog(t *testing.T) {
-	tmpFile, err := os.CreateTemp("", "transformer_compare_log.*.txt")
+	tmpFile, err := os.CreateTemp("", "transformer_compare_log.*.txt.gz")
 	require.NoError(t, err)
 	require.NoError(t, tmpFile.Close())
 
@@ -691,11 +691,12 @@ func TestTransformer_CompareAndLog(t *testing.T) {
 						"event": "track" + strconv.Itoa(index+i+1),
 					},
 					Metadata: ptrans.Metadata{
-						MessageID:       strconv.Itoa(index + i + 1),
-						SourceID:        "sourceID",
-						DestinationID:   "destinationID",
-						SourceType:      "sourceType",
-						DestinationType: "destinationType",
+						MessageID:            strconv.Itoa(index + i + 1),
+						SourceID:             "sourceID",
+						DestinationID:        "destinationID",
+						SourceType:           "sourceType",
+						DestinationType:      "destinationType",
+						SourceDefinitionType: "sourceDefinitionType",
 					},
 				}
 			}),
@@ -714,6 +715,9 @@ func TestTransformer_CompareAndLog(t *testing.T) {
 	require.NoError(t, f.Close())
 
 	differingEvents := strings.Split(strings.Trim(string(data), "\n"), "\n")
+	differingEvents = lo.Filter(differingEvents, func(item string, index int) bool {
+		return strings.Contains(item, "message") // Filtering raw events as the file contains sample diff as well
+	})
 	require.Len(t, differingEvents, maxLoggedEvents)
 
 	for i := 0; i < maxLoggedEvents; i++ {

warehouse/transformer/types.go

@@ -20,6 +20,7 @@ type (
 
 		stats struct {
 			comparisionTime  stats.Timer
+			matchedEvents    stats.Histogram
 			mismatchedEvents stats.Histogram
 		}
 		config struct {