fix: draft event name and product metric

Author: Muhammad Fadli Gunardi

config/dedup.go

@@ -20,8 +20,6 @@ type dedupConfig struct {
 	// Enabled controls whether deduplication is active.
 	// Set DEDUP_ENABLED=true to enable.
 	Enabled bool
-	// IdentifierMapping holds the user and session ID field mappings for a connection group.
-	IdentifierMapping map[string]Identifier
 	// ProtoClassNameMapping maps event_type to proto class name.
 	ProtoClassNameMapping map[string]string
 	// WhitelistConnGroup is a list of connection groups that are processed with dedup.
@@ -31,14 +29,6 @@ type dedupConfig struct {
 func dedupConfigLoader() {
 	viper.SetDefault("DEDUP_ENABLED", "false")
 
-	connGroupIdentifierMap := make(map[string]Identifier)
-	rawMapping := util.MustGetString("DEDUP_IDENTIFIER_MAPPING")
-	if rawMapping != "" {
-		if err := json.Unmarshal([]byte(rawMapping), &connGroupIdentifierMap); err != nil {
-			panic("config: invalid DEDUP_IDENTIFIER_MAPPING: " + err.Error())
-		}
-	}
-
 	var rawWhitelist []string
 
 	rawWhitelistStr := util.MustGetString("DEDUP_WHITELIST_CONN_GROUP")
@@ -62,7 +52,6 @@ func dedupConfigLoader() {
 	DedupCfg = dedupConfig{
 		Enabled:               util.MustGetBool("DEDUP_ENABLED"),
 		WhitelistConnGroup:    whitelistMap,
-		IdentifierMapping:     connGroupIdentifierMap,
 		ProtoClassNameMapping: protoClassNameMap,
 	}
 }

ingestionrule/action/deactivate.go

@@ -7,6 +7,7 @@ import (
 
 	pb "buf.build/gen/go/gotocompany/proton/protocolbuffers/go/gotocompany/raccoon/v1beta1"
 	"github.com/goto/raccoon/config"
+	"github.com/goto/raccoon/ingestionrule/action/dedup/schemaregistry"
 	"github.com/goto/raccoon/ingestionrule/action/eval/cache"
 	"github.com/goto/raccoon/logger"
 	"github.com/goto/raccoon/metrics"
@@ -17,11 +18,16 @@ import (
 type Deactivate struct {
 	cache     *cache.Cache
 	evalChain Chain
+	stencil   schemaregistry.StencilClient
 }
 
 // NewDeactivate creates a new Deactivate action with the given cache and evaluator chain.
-func NewDeactivate(c *cache.Cache, evalChain Chain) *Deactivate {
-	return &Deactivate{cache: c, evalChain: evalChain}
+func NewDeactivate(c *cache.Cache, evalChain Chain, stencil schemaregistry.StencilClient) *Deactivate {
+	return &Deactivate{
+		cache:     c,
+		evalChain: evalChain,
+		stencil:   stencil,
+	}
 }
 
 // Apply evaluates every event in the batch against the deactivate policy rules.
@@ -31,7 +37,12 @@ func (d *Deactivate) Apply(_ context.Context, events []*pb.Event, connGroup stri
 	filtered := make([]*pb.Event, 0, len(events))
 
 	for _, event := range events {
-		meta := ExtractMetadata(event, connGroup, config.PolicyCfg.PublisherMapping, config.EventDistribution.PublisherPattern)
+		meta, err := ExtractMetadata(event, connGroup, config.PolicyCfg.PublisherMapping, config.EventDistribution.PublisherPattern, d.stencil)
+		if err != nil {
+			logger.Errorf("deactivate: failed to extract metadata: %v", err)
+			metrics.Increment(metricNameEventDeserializationError, fmt.Sprintf("conn_group=%s,reason=%s,event_type=%s,product=%s,event_name=%s", connGroup, getErrorReason(err), event.Type, event.Product, event.EventName))
+		}
+
 		logger.Debugf("[deactivate.Apply] meta: event_name=%s, product=%s, publisher=%s, topic=%s, conn_group=%s", meta.EventName, meta.Product, meta.Publisher, meta.TopicName, meta.ConnGroup)
 
 		if d.evalChain.Run(meta, d.cache) {

ingestionrule/action/deactivate_test.go

@@ -8,8 +8,10 @@ import (
 	pb "buf.build/gen/go/gotocompany/proton/protocolbuffers/go/gotocompany/raccoon/v1beta1"
 	"github.com/goto/raccoon/config"
 	"github.com/goto/raccoon/ingestionrule/action"
+	"github.com/goto/raccoon/ingestionrule/action/dedup/schemaregistry"
 	"github.com/goto/raccoon/ingestionrule/action/eval/cache"
 	"github.com/stretchr/testify/assert"
+	"google.golang.org/protobuf/reflect/protoreflect"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
@@ -24,7 +26,7 @@ func buildDeactivateEventCache(name, product, publisher string) *cache.Cache {
 }
 
 func newDeactivate(c *cache.Cache) *action.Deactivate {
-	return action.NewDeactivate(c, action.DefaultChain())
+	return action.NewDeactivate(c, action.DefaultChain(), schemaregistry.StencilClient{})
 }
 
 func TestDeactivate_DropsMatchingEvent(t *testing.T) {
@@ -92,3 +94,52 @@ func TestDeactivate_DropsMatchingTopicRule(t *testing.T) {
 	}}
 	assert.Empty(t, newDeactivate(c).Apply(context.Background(), events, "pub-a"))
 }
+
+func TestDeactivate_UsesDeserializedPayload(t *testing.T) {
+	config.DedupCfg.ProtoClassNameMapping = map[string]string{
+		"component": "ClickEventProto",
+	}
+
+	metaMsg := &mockMessage{
+		fields: map[string]any{
+			"event_guid": "test-guid-123",
+		},
+	}
+
+	staleTime := time.Now()
+	tsMsg := &mockMessage{
+		fullName: "google.protobuf.Timestamp",
+		fields: map[string]any{
+			"seconds": staleTime.Unix(),
+			"nanos":   int32(staleTime.Nanosecond()),
+		},
+	}
+
+	parsedMsg := &mockMessage{
+		fields: map[string]any{
+			"meta":            metaMsg,
+			"event_name":      "deserialized-click",
+			"product":         protoreflect.EnumNumber(1),
+			"event_timestamp": tsMsg,
+		},
+	}
+
+	ms := &mockStencilClient{
+		parseFunc: func(className string, data []byte) (protoreflect.ProtoMessage, error) {
+			return parsedMsg, nil
+		},
+	}
+
+	c := buildDeactivateEventCache("deserialized-click", "clickstream", "pub-a")
+	d := action.NewDeactivate(c, action.DefaultChain(), schemaregistry.StencilClient{Client: ms})
+
+	events := []*pb.Event{{
+		Type:           "component",
+		EventName:      "wrapper-click",
+		Product:        "wrapper-app",
+		EventBytes:     []byte("some-bytes"),
+		EventTimestamp: timestamppb.New(time.Now()),
+	}}
+
+	assert.Empty(t, d.Apply(context.Background(), events, "pub-a"))
+}

ingestionrule/action/dedup.go

@@ -3,45 +3,17 @@ package action
 import (
 	"context"
 	"fmt"
-	"strings"
 	"time"
 
 	pb "buf.build/gen/go/gotocompany/proton/protocolbuffers/go/gotocompany/raccoon/v1beta1"
-	"github.com/spf13/cast"
-	"google.golang.org/protobuf/reflect/protoreflect"
 
 	"github.com/goto/raccoon/config"
 	"github.com/goto/raccoon/ingestionrule/action/dedup/cache"
-	"github.com/goto/raccoon/ingestionrule/action/dedup/protoutil"
 	"github.com/goto/raccoon/ingestionrule/action/dedup/schemaregistry"
 	"github.com/goto/raccoon/logger"
 	"github.com/goto/raccoon/metrics"
 )
 
-const (
-	metricNameEventDeserializationError    = "event_deserialization_error"
-	metricNameEventDeserializationLatency  = "event_deserialization_latency"
-	metricNameEventDuplicateCheckerLatency = "event_duplicate_checker_latency"
-)
-
-const (
-	reasonProtoClassNotFound = "proto class not found"
-	reasonStencilParseError  = "stencil parse error"
-	reasonPublisherNotFound  = "publisher not found"
-
-	reasonUserIDNotFound    = "userID not found"
-	reasonUserIDTypeInvalid = "userID type invalid"
-
-	reasonSessionIDNotFound    = "sessionID not found"
-	reasonSessionIDTypeInvalid = "sessionID type invalid"
-
-	reasonEventNameNotFound    = "event_name not found"
-	reasonEventNameTypeInvalid = "event_name type invalid"
-
-	reasonEventTimestampNotFound    = "event_timestamp not found"
-	reasonEventTimestampTypeInvalid = "event_timestamp type invalid"
-)
-
 // DuplicateChecker defines the capability to verify event uniqueness.
 type DuplicateChecker interface {
 	AreDuplicates(ctx context.Context, events []cache.EventMetadata) ([]bool, error)
@@ -51,8 +23,6 @@ type DuplicateChecker interface {
 
 // processState holds the state of each event being processed.
 type processState struct {
-	// event is the original event being processed.
-	event *pb.Event
 	// isValid indicates whether the event has valid metadata and should be checked for duplication.
 	isValid bool
 }
@@ -91,23 +61,29 @@ func (d *Dedup) Apply(ctx context.Context, events []*pb.Event, connGroup string)
 
 	for i, event := range events {
 		startDeserialize := time.Now()
-		meta, err := d.extractMetadata(event, connGroup)
+		meta, err := ExtractMetadata(event, connGroup, config.PolicyCfg.PublisherMapping, config.EventDistribution.PublisherPattern, d.stencil)
 		metrics.Timing(metricNameEventDeserializationLatency, time.Since(startDeserialize).Milliseconds(), fmt.Sprintf("conn_group=%s", connGroup))
 
 		if err != nil {
 			logger.Errorf("dedup: failed to extract metadata: %v", err)
-			states[i] = processState{event: event, isValid: false}
+			metrics.Increment(metricNameEventDeserializationError, fmt.Sprintf("conn_group=%s,reason=%s,event_type=%s,product=%s,event_name=%s", connGroup, getErrorReason(err), event.Type, event.Product, event.EventName))
+			states[i] = processState{isValid: false}
 			continue
 		}
 
 		if meta.EventGUID == "" || meta.Publisher == "" {
 			logger.Errorf("dedup: missing metadata fields: %+v for conn_group=%s,product=%s,event_name=%s", meta, connGroup, event.Product, event.EventName)
-			states[i] = processState{event: event, isValid: false}
+			states[i] = processState{isValid: false}
 			continue
 		}
 
-		states[i] = processState{event: event, isValid: true}
-		metadataBatch = append(metadataBatch, meta)
+		states[i] = processState{isValid: true}
+		metadataBatch = append(metadataBatch, cache.EventMetadata{
+			Publisher: meta.Publisher,
+			EventGUID: meta.EventGUID,
+			EventName: meta.EventName,
+			Product:   meta.Product,
+		})
 	}
 
 	var isDuplicateResults []bool
@@ -122,9 +98,10 @@ func (d *Dedup) Apply(ctx context.Context, events []*pb.Event, connGroup string)
 	uniqueEvents := make([]*pb.Event, 0, len(events))
 	resultIdx := 0 // Tracks our position in the isDuplicateResults slice
 
-	for _, state := range states {
+	for i, state := range states {
+		event := events[i]
 		if !state.isValid {
-			uniqueEvents = append(uniqueEvents, state.event)
+			uniqueEvents = append(uniqueEvents, event)
 			continue
 		}
 
@@ -133,86 +110,22 @@ func (d *Dedup) Apply(ctx context.Context, events []*pb.Event, connGroup string)
 				logger.Errorf("dedup: cache batch verification failed, bypassing filter: %v", cacheErr)
 			}
 
-			uniqueEvents = append(uniqueEvents, state.event)
+			uniqueEvents = append(uniqueEvents, event)
 			resultIdx++
 			continue
 		}
 
 		isDuplicate := isDuplicateResults[resultIdx]
+		meta := metadataBatch[resultIdx]
 		resultIdx++
 
 		if isDuplicate {
-			metrics.Increment(metricEventLossCount, fmt.Sprintf("reason=DEDUP_POLICY,event_name=%s,product=%s,conn_group=%s,event_type=%s", state.event.EventName, state.event.Product, connGroup, state.event.Type))
+			metrics.Increment(metricEventLossCount, fmt.Sprintf("reason=DEDUP_POLICY,event_name=%s,product=%s,conn_group=%s,event_type=%s", meta.EventName, meta.Product, connGroup, event.Type))
 			continue
 		}
 
-		uniqueEvents = append(uniqueEvents, state.event)
+		uniqueEvents = append(uniqueEvents, event)
 	}
 
 	return uniqueEvents
 }
-
-// extractMetadata deserializes dynamic protobuf payloads using Stencil and handles identity field extractions.
-func (d *Dedup) extractMetadata(event *pb.Event, connGroup string) (cache.EventMetadata, error) {
-	protoClass, ok := config.DedupCfg.ProtoClassNameMapping[event.Type]
-	if !ok {
-		metrics.Increment(metricNameEventDeserializationError,
-			fmt.Sprintf("conn_group=%s,reason=%s,event_type=%s,product=%s,event_name=%s", connGroup, reasonProtoClassNotFound, event.Type, event.Product, event.EventName))
-		return cache.EventMetadata{}, fmt.Errorf("failed to find proto class for conn_group=%s,event_type=%s,product=%s,event_name=%s", connGroup, event.Type, event.Product, event.EventName)
-	}
-
-	publisher, ok := config.PolicyCfg.PublisherMapping[connGroup]
-	if !ok {
-		metrics.Increment(metricNameEventDeserializationError,
-			fmt.Sprintf("conn_group=%s,reason=%s,event_type=%s,product=%s,event_name=%s", connGroup, reasonPublisherNotFound, event.Type, event.Product, event.EventName))
-		return cache.EventMetadata{}, fmt.Errorf("failed to publisher for conn_group=%s,event_type=%s,product=%s,event_name=%s", connGroup, event.Type, event.Product, event.EventName)
-	}
-
-	parsedMsg, err := d.stencil.Client.Parse(protoClass, event.EventBytes)
-	if err != nil {
-		metrics.Increment(metricNameEventDeserializationError,
-			fmt.Sprintf("conn_group=%s,reason=%s,event_type=%s,product=%s,event_name=%s", connGroup, reasonStencilParseError, event.Type, event.Product, event.EventName))
-		return cache.EventMetadata{}, fmt.Errorf("failed to parse proto class for conn_group=%s,event_type=%s,product=%s,event_name=%s", connGroup, event.Type, event.Product, event.EventName)
-	}
-
-	ref := parsedMsg.ProtoReflect()
-
-	const protoFieldEventGUID = "meta.event_guid"
-
-	eventGUID, err := d.getStringField(ref, protoFieldEventGUID, connGroup, event, protoFieldEventGUID, reasonEventNameNotFound, reasonEventNameTypeInvalid)
-	if err != nil {
-		return cache.EventMetadata{}, err
-	}
-
-	return cache.EventMetadata{
-		EventGUID: eventGUID,
-		Publisher: publisher,
-	}, nil
-}
-
-// getStringField is a helper function to safely extract, convert to string, and handle error telemetry for identifier fields.
-func (d *Dedup) getStringField(
-	ref protoreflect.Message,
-	path string,
-	connGroup string,
-	event *pb.Event,
-	fieldName string,
-	reasonNotFound string,
-	reasonTypeInvalid string,
-) (string, error) {
-	rawVal, ok := protoutil.GetFieldValue(ref, strings.Split(path, "."))
-	if !ok {
-		metrics.Increment(metricNameEventDeserializationError,
-			fmt.Sprintf("conn_group=%s,reason=%s,event_type=%s,product=%s,event_name=%s", connGroup, reasonNotFound, event.Type, event.Product, event.EventName))
-		return "", fmt.Errorf("failed to find %s for conn_group=%s,event_type=%s,product=%s,event_name=%s", fieldName, connGroup, event.Type, event.Product, event.EventName)
-	}
-
-	val, err := cast.ToStringE(rawVal)
-	if err != nil {
-		metrics.Increment(metricNameEventDeserializationError,
-			fmt.Sprintf("conn_group=%s,reason=%s,event_type=%s,product=%s,event_name=%s", connGroup, reasonTypeInvalid, event.Type, event.Product, event.EventName))
-		return "", fmt.Errorf("%s field type is not convertible to string for conn_group=%s,event_type=%s,product=%s,event_name=%s: %w", fieldName, connGroup, event.Type, event.Product, event.EventName, err)
-	}
-
-	return val, nil
-}

ingestionrule/action/dedup/cache/cache.go

@@ -40,6 +40,8 @@ type Store struct {
 type EventMetadata struct {
 	Publisher string
 	EventGUID string
+	EventName string
+	Product   string
 }
 
 // NewStore instantiates the unified storage framework wrapper.