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

[SDKS-9120] Toggle impressions

Author: sanzmauro

CHANGES

@@ -1,3 +1,6 @@
+1.5.0 (Jan 17, 2025):
+- Added support for the new impressions tracking toggle available on feature flags, both respecting the setting and including the new field being returned on SplitView type objects. Read more in our docs.
+
 1.4.0 (May 14, 2024):
 - Updated go-split-commons to v6
   - Added support for targeting rules based on semantic versions (https://semver.org/).

external/commons/mocks/impmanager.go

@@ -22,4 +22,9 @@ func (m *ImpressionManagerMock) ProcessSingle(impression *dtos.Impression) bool
 	return args.Bool(0)
 }
 
+func (m *ImpressionManagerMock) Process(impressions []dtos.Impression, listenerEnabled bool) ([]dtos.Impression, []dtos.Impression) {
+	args := m.Called(impressions)
+	return args.Get(0).([]dtos.Impression), args.Get(1).([]dtos.Impression)
+}
+
 var _ provisional.ImpressionManager = (*ImpressionManagerMock)(nil)

external/commons/mocks/splitstorage.go

@@ -68,4 +68,8 @@ func (m *SplitStorageMock) GetAllFlagSetNames() []string {
 	return args.Get(0).([]string)
 }
 
+func (m *SplitStorageMock) LargeSegmentNames() *set.ThreadUnsafeSet {
+	return m.Called().Get(0).(*set.ThreadUnsafeSet)
+}
+
 var _ storage.SplitStorage = (*SplitStorageMock)(nil)

go.mod

@@ -3,18 +3,20 @@ module github.com/splitio/splitd
 go 1.21
 
 require (
-	github.com/splitio/go-split-commons/v6 v6.0.0
+	github.com/splitio/go-split-commons/v6 v6.1.0
 	github.com/splitio/go-toolkit/v5 v5.4.0
-	github.com/stretchr/testify v1.8.1
+	github.com/stretchr/testify v1.9.0
 	github.com/vmihailenco/msgpack/v5 v5.3.5
 	golang.org/x/sync v0.3.0
 	gopkg.in/yaml.v3 v3.0.1
 )
 
 require (
+	github.com/bits-and-blooms/bitset v1.3.1 // indirect
+	github.com/bits-and-blooms/bloom/v3 v3.3.1 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/stretchr/objx v0.5.0 // indirect
+	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
 	golang.org/x/exp v0.0.0-20231006140011-7918f672742d // indirect
 )

go.sum

@@ -7,19 +7,18 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/splitio/go-split-commons/v6 v6.0.0 h1:qenr5qbXafjvM832C64CVpjtlShuQiWCwtR5I2h4ogM=
-github.com/splitio/go-split-commons/v6 v6.0.0/go.mod h1:TsvIh3XP7yjc7ly4vpj06AkoBND36SodPs5qfhb8rHc=
+github.com/splitio/go-split-commons/v6 v6.1.0 h1:k3mwr12DF6gbEaV8XXU/tSAQlPkIEuzIgTEneYhGg2I=
+github.com/splitio/go-split-commons/v6 v6.1.0/go.mod h1:D/XIY/9Hmfk9ivWsRsJVp439kEdmHbzUi3PKzQQDOXY=
 github.com/splitio/go-toolkit/v5 v5.4.0 h1:g5WFpRhQomnXCmvfsNOWV4s5AuUrWIZ+amM68G8NBKM=
 github.com/splitio/go-toolkit/v5 v5.4.0/go.mod h1:xYhUvV1gga9/1029Wbp5pjnR6Cy8nvBpjw99wAbsMko=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
-github.com/stretchr/objx v0.5.0 h1:1zr/of2m5FGMsad5YfcqgdqdWrIhu+EBEJRhR1U7z/c=
-github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
+github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY=
+github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
-github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
-github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
+github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
+github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/twmb/murmur3 v1.1.6 h1:mqrRot1BRxm+Yct+vavLMou2/iJt0tNVTTC0QoIjaZg=
+github.com/twmb/murmur3 v1.1.6/go.mod h1:Qq/R7NUyOfr65zD+6Q5IHKsJLwP7exErjN6lyyq3OSQ=
 github.com/vmihailenco/msgpack/v5 v5.3.5 h1:5gO0H1iULLWGhs2H5tbAHIZTV8/cYafcFOr9znI5mJU=
 github.com/vmihailenco/msgpack/v5 v5.3.5/go.mod h1:7xyJ9e+0+9SaZT0Wt1RGleJXzli6Q/V5KbhBonMG9jc=
 github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g=

splitio/link/protocol/v1/responses.go

@@ -37,14 +37,15 @@ type SplitNamesPayload struct {
 }
 
 type SplitPayload struct {
-	Name             string            `msgpack:"n"`
-	TrafficType      string            `msgpack:"t"`
-	Killed           bool              `msgpack:"k"`
-	Treatments       []string          `msgpack:"s"`
-	ChangeNumber     int64             `msgpack:"c"`
-	Configs          map[string]string `msgpack:"f"`
-	DefaultTreatment string            `msgpack:"d"`
-	Sets             []string          `msgpack:"e"`
+	Name                string            `msgpack:"n"`
+	TrafficType         string            `msgpack:"t"`
+	Killed              bool              `msgpack:"k"`
+	Treatments          []string          `msgpack:"s"`
+	ChangeNumber        int64             `msgpack:"c"`
+	Configs             map[string]string `msgpack:"f"`
+	DefaultTreatment    string            `msgpack:"d"`
+	Sets                []string          `msgpack:"e"`
+	ImpressionsDisabled bool              `msgpack:"i"`
 }
 
 type SplitsPayload struct {

splitio/sdk/helpers.go

@@ -15,43 +15,58 @@ import (
 	"github.com/splitio/go-split-commons/v6/provisional/strategy"
 	"github.com/splitio/go-split-commons/v6/service/api"
 	"github.com/splitio/go-split-commons/v6/storage"
+	"github.com/splitio/go-split-commons/v6/storage/filter"
 	"github.com/splitio/go-split-commons/v6/storage/inmemory"
 	"github.com/splitio/go-split-commons/v6/storage/inmemory/mutexmap"
 	"github.com/splitio/go-split-commons/v6/synchronizer"
 	"github.com/splitio/go-split-commons/v6/synchronizer/worker/impressionscount"
 	"github.com/splitio/go-split-commons/v6/synchronizer/worker/segment"
 	"github.com/splitio/go-split-commons/v6/synchronizer/worker/split"
 	"github.com/splitio/go-split-commons/v6/tasks"
+	"github.com/splitio/go-split-commons/v6/telemetry"
 	"github.com/splitio/go-toolkit/v5/logging"
 )
 
+const (
+	bfExpectedElemenets                = 10000000
+	bfFalsePositiveProbability         = 0.01
+	bfCleaningPeriod                   = 86400 // 24 hours
+	uniqueKeysPeriodTaskInMemory       = 900   // 15 min
+	uniqueKeysPeriodTaskRedis          = 300   // 5 min
+	impressionsCountPeriodTaskInMemory = 1800  // 30 min
+	impressionsCountPeriodTaskRedis    = 300   // 5 min
+	impressionsBulkSizeRedis           = 100
+)
+
 func setupWorkers(
 	logger logging.LoggerInterface,
 	api *api.SplitAPI,
 	str *storages,
 	hc application.MonitorProducerInterface,
 	cfg *sdkConf.Config,
 	flagSetsFilter flagsets.FlagSetFilter,
+	md dtos.Metadata,
+	impComponents impComponents,
 ) *synchronizer.Workers {
 	return &synchronizer.Workers{
-		SplitUpdater:       split.NewSplitUpdater(str.splits, api.SplitFetcher, logger, str.telemetry, hc, flagSetsFilter),
-		SegmentUpdater:     segment.NewSegmentUpdater(str.splits, str.segments, api.SegmentFetcher, logger, str.telemetry, hc),
-		ImpressionRecorder: workers.NewImpressionsWorker(logger, str.telemetry, api.ImpressionRecorder, str.impressions, &cfg.Impressions),
-		EventRecorder:      workers.NewEventsWorker(logger, str.telemetry, api.EventRecorder, str.events, &cfg.Events),
+		SplitUpdater:             split.NewSplitUpdater(str.splits, api.SplitFetcher, logger, str.telemetry, hc, flagSetsFilter),
+		SegmentUpdater:           segment.NewSegmentUpdater(str.splits, str.segments, api.SegmentFetcher, logger, str.telemetry, hc),
+		ImpressionRecorder:       workers.NewImpressionsWorker(logger, str.telemetry, api.ImpressionRecorder, str.impressions, &cfg.Impressions),
+		EventRecorder:            workers.NewEventsWorker(logger, str.telemetry, api.EventRecorder, str.events, &cfg.Events),
+		ImpressionsCountRecorder: impressionscount.NewRecorderSingle(impComponents.counter, api.ImpressionRecorder, md, logger, str.telemetry),
+		TelemetryRecorder:        telemetry.NewTelemetrySynchronizer(str.telemetry, api.TelemetryRecorder, str.splits, str.segments, logger, md, str.telemetry),
 	}
 }
 
 func setupTasks(
 	cfg *sdkConf.Config,
-	str *storages,
 	logger logging.LoggerInterface,
 	workers *synchronizer.Workers,
 	impComponents impComponents,
-	md dtos.Metadata,
-	api *api.SplitAPI,
 ) *synchronizer.SplitTasks {
 	impCfg := cfg.Impressions
 	evCfg := cfg.Events
+	dummyHC := &application.Dummy{}
 	tg := &synchronizer.SplitTasks{
 		SplitSyncTask: tasks.NewFetchSplitsTask(workers.SplitUpdater, int(cfg.Splits.SyncPeriod.Seconds()), logger),
 		SegmentSyncTask: tasks.NewFetchSegmentsTask(
@@ -60,21 +75,18 @@ func setupTasks(
 			cfg.Segments.WorkerCount,
 			cfg.Segments.QueueSize,
 			logger,
+			dummyHC,
 		),
-		ImpressionSyncTask:    tasks.NewRecordImpressionsTask(workers.ImpressionRecorder, int(impCfg.SyncPeriod.Seconds()), logger, 5000),
-		EventSyncTask:         tasks.NewRecordEventsTask(workers.EventRecorder, 5000, int(evCfg.SyncPeriod.Seconds()), logger),
-		TelemetrySyncTask:     &NoOpTask{},
-		UniqueKeysTask:        &NoOpTask{},
-		CleanFilterTask:       &NoOpTask{},
-		ImpsCountConsumerTask: &NoOpTask{},
-	}
-
-	if impCfg.Mode == "optimized" {
-		tg.ImpressionsCountSyncTask = tasks.NewRecordImpressionsCountTask(
-			impressionscount.NewRecorderSingle(impComponents.counter, api.ImpressionRecorder, md, logger, str.telemetry),
+		ImpressionSyncTask: tasks.NewRecordImpressionsTask(workers.ImpressionRecorder, int(impCfg.SyncPeriod.Seconds()), logger, 5000),
+		EventSyncTask:      tasks.NewRecordEventsTask(workers.EventRecorder, 5000, int(evCfg.SyncPeriod.Seconds()), logger),
+		TelemetrySyncTask:  &NoOpTask{},
+		UniqueKeysTask:     tasks.NewRecordUniqueKeysTask(workers.TelemetryRecorder, *impComponents.tracker, uniqueKeysPeriodTaskInMemory, logger),
+		CleanFilterTask:    tasks.NewCleanFilterTask(*impComponents.filter, logger, bfCleaningPeriod),
+		ImpsCountConsumerTask: tasks.NewRecordImpressionsCountTask(
+			workers.ImpressionsCountRecorder,
 			logger,
 			int(impCfg.CountSyncPeriod.Seconds()),
-		)
+		),
 	}
 
 	return tg
@@ -83,6 +95,8 @@ func setupTasks(
 type impComponents struct {
 	manager provisional.ImpressionManager
 	counter *strategy.ImpressionsCounter
+	tracker *strategy.UniqueKeysTracker
+	filter  *storage.Filter
 }
 
 func setupImpressionsComponents(c *sdkConf.Impressions, telemetry storage.TelemetryRuntimeProducer) (impComponents, error) {
@@ -92,20 +106,28 @@ func setupImpressionsComponents(c *sdkConf.Impressions, telemetry storage.Teleme
 		return impComponents{}, fmt.Errorf("error building impressions observer: %w", err)
 	}
 
+	counter := strategy.NewImpressionsCounter()
+	bf := filter.NewBloomFilter(bfExpectedElemenets, bfFalsePositiveProbability)
+	tracker := strategy.NewUniqueKeysTracker(bf)
+	none := strategy.NewNoneImpl(counter, tracker, false)
+
 	var s strategy.ProcessStrategyInterface
-	var counter *strategy.ImpressionsCounter
 	switch c.Mode {
 	case conf.ImpressionsModeDebug:
 		s = strategy.NewDebugImpl(observer, false)
 	case conf.ImpressionsModeNone:
+		s = none
 	default: // optimized
-		counter = strategy.NewImpressionsCounter()
 		s = strategy.NewOptimizedImpl(observer, counter, telemetry, false)
 	}
 
+	impManager := provisional.NewImpressionManagerImp(none, s)
+
 	return impComponents{
-		manager: provisional.NewImpressionManager(s),
+		manager: impManager,
 		counter: counter,
+		tracker: &tracker,
+		filter:  &bf,
 	}, nil
 }
 

splitio/sdk/helpers_test.go

@@ -16,11 +16,15 @@ func TestSetupImpressionsComponents(t *testing.T) {
 	ic, err := setupImpressionsComponents(&sdkCfg.Impressions, storages.telemetry)
 	assert.Nil(t, err)
 	assert.NotNil(t, ic.counter)
+	assert.NotNil(t, ic.tracker)
+	assert.NotNil(t, ic.filter)
 
 	sdkCfg.Impressions.Mode = "debug"
 	ic, err = setupImpressionsComponents(&sdkCfg.Impressions, storages.telemetry)
 	assert.Nil(t, err)
-	assert.Nil(t, ic.counter)
+	assert.NotNil(t, ic.counter)
+	assert.NotNil(t, ic.tracker)
+	assert.NotNil(t, ic.filter)
 }
 
 func TestNoOpTask(t *testing.T) {

splitio/sdk/results.go

@@ -9,12 +9,13 @@ type EvaluationResult struct {
 }
 
 type SplitView struct {
-	Name             string
-	TrafficType      string
-	Killed           bool
-	Treatments       []string
-	ChangeNumber     int64
-	Configs          map[string]string
-	DefaultTreatment string
-	Sets             []string
+	Name                string
+	TrafficType         string
+	Killed              bool
+	Treatments          []string
+	ChangeNumber        int64
+	Configs             map[string]string
+	DefaultTreatment    string
+	Sets                []string
+	ImpressionsDisabled bool
 }

splitio/sdk/sdk.go

@@ -84,8 +84,8 @@ func New(logger logging.LoggerInterface, apikey string, c *conf.Config) (*Impl,
 
 	queueFullChan := make(chan string, 2)
 	splitApi := api.NewSplitAPI(apikey, *advCfg, logger, md)
-	workers := setupWorkers(logger, splitApi, stores, hc, c, flagSetsFilter)
-	tasks := setupTasks(c, stores, logger, workers, impc, md, splitApi)
+	workers := setupWorkers(logger, splitApi, stores, hc, c, flagSetsFilter, md, impc)
+	tasks := setupTasks(c, logger, workers, impc)
 	sync := synchronizer.NewSynchronizer(*advCfg, *tasks, *workers, logger, queueFullChan)
 
 	status := make(chan int, 10)
@@ -257,19 +257,20 @@ func (i *Impl) handleImpression(key string, bk *string, f string, r *evaluator.R
 		label = r.Label
 	}
 
-	imp := &dtos.Impression{
+	imp := dtos.Impression{
 		FeatureName:  f,
 		BucketingKey: common.StringFromRef(bk),
 		ChangeNumber: r.SplitChangeNumber,
 		KeyName:      key,
 		Label:        label,
 		Treatment:    r.Treatment,
 		Time:         timeMillis(),
+		Disabled:     r.ImpressionsDisabled,
 	}
 
-	shouldStore := i.iq.ProcessSingle(imp)
-	if shouldStore {
-		_, err := i.is.Push(cm, *imp)
+	forLog, _ := i.iq.Process([]dtos.Impression{imp}, false)
+	if len(forLog) == 1 {
+		_, err := i.is.Push(cm, forLog[0])
 		if err != nil {
 			if err == storage.ErrQueueFull {
 				select {
@@ -283,7 +284,7 @@ func (i *Impl) handleImpression(key string, bk *string, f string, r *evaluator.R
 		}
 	}
 
-	return imp
+	return &imp
 }
 
 func splitToView(s *dtos.SplitDTO) *SplitView {
@@ -296,14 +297,15 @@ func splitToView(s *dtos.SplitDTO) *SplitView {
 	}
 
 	return &SplitView{
-		Name:             s.Name,
-		TrafficType:      s.TrafficTypeName,
-		Killed:           s.Killed,
-		ChangeNumber:     s.ChangeNumber,
-		Configs:          s.Configurations,
-		Treatments:       treatments,
-		DefaultTreatment: s.DefaultTreatment,
-		Sets:             s.Sets,
+		Name:                s.Name,
+		TrafficType:         s.TrafficTypeName,
+		Killed:              s.Killed,
+		ChangeNumber:        s.ChangeNumber,
+		Configs:             s.Configurations,
+		Treatments:          treatments,
+		DefaultTreatment:    s.DefaultTreatment,
+		Sets:                s.Sets,
+		ImpressionsDisabled: s.ImpressionsDisabled,
 	}
 }