Merge pull request #202 from splitio/SDKS-9120-impressoion-per-toggle

[SDKS-9120] Impression per toggle implementation
splitio · Jan 14, 2025 · 07f0338 · 07f0338
2 parents 678a6e7 + 1d29216
commit 07f0338
Show file tree

Hide file tree

Showing 8 changed files with 117 additions and 16 deletions.
diff --git a/dtos/impression.go b/dtos/impression.go
@@ -10,6 +10,7 @@ type Impression struct {
 	ChangeNumber int64  `json:"c"`
 	Time         int64  `json:"m"`
 	Pt           int64  `json:"pt,omitempty"`
+	Disabled     bool   `json:"-"`
 }
 
 // ImpressionQueueObject struct mapping impressions

diff --git a/dtos/split.go b/dtos/split.go
@@ -24,6 +24,7 @@ type SplitDTO struct {
 	Conditions            []ConditionDTO    `json:"conditions"`
 	Configurations        map[string]string `json:"configurations"`
 	Sets                  []string          `json:"sets"`
+	ImpressionsDisabled   bool              `json:"impressionsDisabled"`
 }
 
 // MarshalBinary exports SplitDTO to JSON string

diff --git a/engine/evaluator/evaluator.go b/engine/evaluator/evaluator.go
@@ -22,11 +22,12 @@ const (
 // Result represents the result of an evaluation, including the resulting treatment, the label for the impression,
 // the latency and error if any
 type Result struct {
-	Treatment         string
-	Label             string
-	EvaluationTime    time.Duration
-	SplitChangeNumber int64
-	Config            *string
+	Treatment           string
+	Label               string
+	EvaluationTime      time.Duration
+	SplitChangeNumber   int64
+	Config              *string
+	ImpressionsDisabled bool
 }
 
 // Results represents the result of multiple evaluations at once
@@ -84,10 +85,11 @@ func (e *Evaluator) evaluateTreatment(key string, bucketingKey string, featureFl
 		}
 
 		return &Result{
-			Treatment:         split.DefaultTreatment(),
-			Label:             impressionlabels.Killed,
-			SplitChangeNumber: split.ChangeNumber(),
-			Config:            config,
+			Treatment:           split.DefaultTreatment(),
+			Label:               impressionlabels.Killed,
+			SplitChangeNumber:   split.ChangeNumber(),
+			Config:              config,
+			ImpressionsDisabled: split.ImpressionsDisabled(),
 		}
 	}
 
@@ -108,10 +110,11 @@ func (e *Evaluator) evaluateTreatment(key string, bucketingKey string, featureFl
 	}
 
 	return &Result{
-		Treatment:         *treatment,
-		Label:             label,
-		SplitChangeNumber: split.ChangeNumber(),
-		Config:            config,
+		Treatment:           *treatment,
+		Label:               label,
+		SplitChangeNumber:   split.ChangeNumber(),
+		Config:              config,
+		ImpressionsDisabled: split.ImpressionsDisabled(),
 	}
 }
 

diff --git a/engine/grammar/split.go b/engine/grammar/split.go
@@ -111,3 +111,7 @@ func (s *Split) ChangeNumber() int64 {
 func (s *Split) Configurations() map[string]string {
 	return s.splitData.Configurations
 }
+
+func (s *Split) ImpressionsDisabled() bool {
+	return s.splitData.ImpressionsDisabled
+}
diff --git a/provisional/impmanager.go b/provisional/impmanager.go
@@ -9,20 +9,30 @@ import (
 type ImpressionManager interface {
 	ProcessImpressions(impressions []dtos.Impression) ([]dtos.Impression, []dtos.Impression)
 	ProcessSingle(impression *dtos.Impression) bool
+	Process(impressions []dtos.Impression, listenerEnabled bool) ([]dtos.Impression, []dtos.Impression)
 }
 
 // ImpressionManagerImpl implements
 type ImpressionManagerImpl struct {
 	processStrategy strategy.ProcessStrategyInterface
+	keyTracker      strategy.ProcessStrategyInterface
 }
 
+// DEPRECATED
 // NewImpressionManager creates new ImpManager
 func NewImpressionManager(processStrategy strategy.ProcessStrategyInterface) ImpressionManager {
 	return &ImpressionManagerImpl{
 		processStrategy: processStrategy,
 	}
 }
 
+func NewImpressionManagerImp(none *strategy.NoneImpl, processStrategy strategy.ProcessStrategyInterface) ImpressionManager {
+	return &ImpressionManagerImpl{
+		processStrategy: processStrategy,
+		keyTracker:      none,
+	}
+}
+
 // ProcessImpressions bulk processes
 func (i *ImpressionManagerImpl) ProcessImpressions(impressions []dtos.Impression) ([]dtos.Impression, []dtos.Impression) {
 	return i.processStrategy.Apply(impressions)
@@ -33,3 +43,22 @@ func (i *ImpressionManagerImpl) ProcessImpressions(impressions []dtos.Impression
 func (i *ImpressionManagerImpl) ProcessSingle(impression *dtos.Impression) bool {
 	return i.processStrategy.ApplySingle(impression)
 }
+
+func (i *ImpressionManagerImpl) Process(impressions []dtos.Impression, listenerEnabled bool) ([]dtos.Impression, []dtos.Impression) {
+	forLog := make([]dtos.Impression, 0, len(impressions))
+	forListener := make([]dtos.Impression, 0, len(impressions))
+
+	for index := range impressions {
+		if impressions[index].Disabled {
+			i.keyTracker.ApplySingle(&impressions[index])
+		} else if i.processStrategy.ApplySingle(&impressions[index]) {
+			forLog = append(forLog, impressions[index])
+		}
+	}
+
+	if listenerEnabled {
+		forListener = impressions
+	}
+
+	return forLog, forListener
+}
diff --git a/provisional/impmanager_test.go b/provisional/impmanager_test.go
@@ -185,3 +185,65 @@ func TestImpManagerRedis(t *testing.T) {
 		t.Error("It should have pt")
 	}
 }
+
+func TestProcess(t *testing.T) {
+	observer, _ := strategy.NewImpressionObserver(5000)
+	debug := strategy.NewDebugImpl(observer, true)
+	filter := filter.NewBloomFilter(3000, 0.01)
+	uniqueTracker := strategy.NewUniqueKeysTracker(filter)
+	counter := strategy.NewImpressionsCounter()
+	none := strategy.NewNoneImpl(counter, uniqueTracker, false)
+
+	now := time.Now().UTC().UnixNano()
+	impressions := []dtos.Impression{
+		{
+			BucketingKey: "someBucketingKey",
+			ChangeNumber: 123456789,
+			FeatureName:  "someFeature",
+			KeyName:      "someKey",
+			Label:        "someLabel",
+			Time:         now,
+			Treatment:    "someTreatment",
+			Disabled:     true,
+		},
+		{
+			BucketingKey: "someBucketingKey",
+			ChangeNumber: 123456789,
+			FeatureName:  "harnessFlag",
+			KeyName:      "someKey",
+			Label:        "someLabel",
+			Time:         now,
+			Treatment:    "someTreatment",
+			Disabled:     true,
+		},
+		{
+			BucketingKey: "someBucketingKey",
+			ChangeNumber: 123456789,
+			FeatureName:  "featureTest",
+			KeyName:      "someKey",
+			Label:        "someLabel",
+			Time:         now,
+			Treatment:    "someTreatment",
+			Disabled:     false,
+		},
+	}
+
+	impManager := NewImpressionManagerImp(none, debug)
+	impressionsForLog, impressionsForListener := impManager.Process(impressions, true)
+	if len(impressionsForListener) != 3 {
+		t.Error("Impressions for Listener should be 3. Actual: ", len(impressionsForListener))
+	}
+	if len(impressionsForLog) != 1 {
+		t.Error("Impressions for Log should be 3. Actual: ", len(impressionsForLog))
+	}
+
+	impManager = NewImpressionManagerImp(none, none)
+
+	impressionsForLog, impressionsForListener = impManager.Process(impressions, false)
+	if len(impressionsForListener) != 0 {
+		t.Error("Impressions for Listener should be 0. Actual: ", len(impressionsForListener))
+	}
+	if len(impressionsForLog) != 0 {
+		t.Error("Impressions for Log should be 1. Actual: ", len(impressionsForLog))
+	}
+}
diff --git a/provisional/strategy/none.go b/provisional/strategy/none.go
@@ -14,7 +14,7 @@ type NoneImpl struct {
 }
 
 // NewNoneImpl creates new NoneImpl.
-func NewNoneImpl(impressionCounter *ImpressionsCounter, uniqueKeysTracker UniqueKeysTracker, listenerEnabled bool) ProcessStrategyInterface {
+func NewNoneImpl(impressionCounter *ImpressionsCounter, uniqueKeysTracker UniqueKeysTracker, listenerEnabled bool) *NoneImpl {
 	return &NoneImpl{
 		impressionsCounter: impressionCounter,
 		uniqueKeysTracker:  uniqueKeysTracker,
@@ -51,3 +51,5 @@ func (s *NoneImpl) ApplySingle(impression *dtos.Impression) bool {
 
 	return s.apply(impression, now)
 }
+
+var _ ProcessStrategyInterface = (*NoneImpl)(nil)
diff --git a/provisional/strategy/optimized.go b/provisional/strategy/optimized.go
@@ -34,6 +34,7 @@ func (s *OptimizedImpl) apply(impression *dtos.Impression, now int64) bool {
 	}
 
 	if impression.Pt == 0 || impression.Pt < util.TruncateTimeFrame(now) {
+		s.runtimeTelemetry.RecordImpressionsStats(telemetry.ImpressionsDeduped, 1)
 		return true
 	}
 
@@ -56,8 +57,6 @@ func (s *OptimizedImpl) Apply(impressions []dtos.Impression) ([]dtos.Impression,
 		forListener = impressions
 	}
 
-	s.runtimeTelemetry.RecordImpressionsStats(telemetry.ImpressionsDeduped, int64(len(impressions)-len(forLog)))
-
 	return forLog, forListener
 }