FFM-11660 Add size and clearing interval to safe seen targets

FFM-11660 Update tests

FFM-11660 Make seen targets max size and clearing schedule configurable with defaults

FFM-11660 Change level of bucketby missing log to debug

FFM-11660 Add size and clearing interval to safe seen targets

Author: erdirowlands

analyticsservice/analytics.go

@@ -46,6 +46,13 @@ type SafeAnalyticsCache[K comparable, V any] interface {
 	iterate(func(K, V))
 }
 
+// SafeSeenTargetsCache extends SafeAnalyticsCache and adds behavior specific to seen targets
+type SafeSeenTargetsCache[K comparable, V any] interface {
+	SafeAnalyticsCache[K, V]
+	setWithLimit(key K, value V)
+	isLimitExceeded() bool
+}
+
 type analyticsEvent struct {
 	target        *evaluation.Target
 	featureConfig *rest.FeatureConfig
@@ -68,7 +75,7 @@ type AnalyticsService struct {
 }
 
 // NewAnalyticsService creates and starts a analytics service to send data to the client
-func NewAnalyticsService(timeout time.Duration, logger logger.Logger) *AnalyticsService {
+func NewAnalyticsService(timeout time.Duration, logger logger.Logger, seenTargetsMaxSize int, seenTargetsClearingSchedule time.Duration) *AnalyticsService {
 	serviceTimeout := timeout
 	if timeout < 60*time.Second {
 		serviceTimeout = 60 * time.Second
@@ -79,7 +86,7 @@ func NewAnalyticsService(timeout time.Duration, logger logger.Logger) *Analytics
 		analyticsChan:       make(chan analyticsEvent),
 		evaluationAnalytics: newSafeEvaluationAnalytics(),
 		targetAnalytics:     newSafeTargetAnalytics(),
-		seenTargets:         newSafeSeenTargets(),
+		seenTargets:         newSafeSeenTargets(seenTargetsMaxSize, seenTargetsClearingSchedule),
 		timeout:             serviceTimeout,
 		logger:              logger,
 	}

analyticsservice/analytics_test.go

@@ -120,7 +120,7 @@ func TestListenerHandlesEventsCorrectly(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			service := NewAnalyticsService(1*time.Minute, noOpLogger)
+			service := NewAnalyticsService(1*time.Minute, noOpLogger, 10, time.Hour)
 			defer close(service.analyticsChan)
 
 			// Start the listener in a goroutine

analyticsservice/safe_maps_test.go

@@ -1,6 +1,7 @@
 package analyticsservice
 
 import (
+	"fmt"
 	"reflect"
 	"sync"
 	"testing"
@@ -81,13 +82,94 @@ func TestSafeTargetAnalytics(t *testing.T) {
 }
 
 func TestSafeSeenTargets(t *testing.T) {
-	s := newSafeSeenTargets()
+	// Initialize with a small maxSize for testing
+	maxSize := 3
+	s := newSafeSeenTargets(maxSize, 0).(SafeSeenTargetsCache[string, bool])
+
 	testData := map[string]bool{
 		"target1":  true,
 		"target21": true,
 		"target3":  true,
-		"target4":  true,
 	}
 
-	testMapOperations[string, bool](t, s, testData)
+	// Insert items and ensure limit is not exceeded
+	for key, value := range testData {
+		s.set(key, value)
+	}
+
+	if s.isLimitExceeded() {
+		t.Errorf("Limit should not have been exceeded yet")
+	}
+
+	// Add one more item to exceed the limit
+	s.setWithLimit("target4", true)
+
+	// Ensure limitExceeded is true after exceeding the limit
+	if !s.isLimitExceeded() {
+		t.Errorf("Limit should be exceeded after adding target4")
+	}
+
+	// Ensure that new items are not added once the limit is exceeded
+	s.setWithLimit("target5", true)
+	if _, exists := s.get("target5"); exists {
+		t.Errorf("target5 should not have been added as the limit was exceeded")
+	}
+
+	// Clear the map and ensure limit is reset
+	s.clear()
+
+	if s.isLimitExceeded() {
+		t.Errorf("Limit should have been reset after clearing the map")
+	}
+
+	// Add items again after clearing
+	s.setWithLimit("target6", true)
+	if _, exists := s.get("target6"); !exists {
+		t.Errorf("target6 should have been added after clearing the map")
+	}
+
+	// Concurrency test
+	t.Run("ConcurrencyTest", func(t *testing.T) {
+		var wg sync.WaitGroup
+		concurrencyLevel := 100
+
+		// Re-initialize the map for concurrency testing
+		s = newSafeSeenTargets(100, 0).(SafeSeenTargetsCache[string, bool])
+
+		// Concurrently set keys
+		for i := 0; i < concurrencyLevel; i++ {
+			wg.Add(1)
+			go func(i int) {
+				defer wg.Done()
+				key := "target" + fmt.Sprint(i)
+				s.setWithLimit(key, true)
+			}(i)
+		}
+
+		// Concurrently get keys
+		for i := 0; i < concurrencyLevel; i++ {
+			wg.Add(1)
+			go func(i int) {
+				defer wg.Done()
+				key := "target" + fmt.Sprint(i)
+				s.get(key)
+			}(i)
+		}
+
+		// Concurrently clear the map
+		for i := 0; i < concurrencyLevel/2; i++ {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				s.clear()
+			}()
+		}
+
+		wg.Wait()
+
+		// Ensure the map is cleared after the concurrency operations
+		if s.size() > 0 {
+			t.Errorf("Map size should be 0 after clearing, got %d", s.size())
+		}
+	})
 }

analyticsservice/safe_seen_targets_map.go

@@ -2,25 +2,59 @@ package analyticsservice
 
 import (
 	"sync"
+	"sync/atomic"
+	"time"
 )
 
 type safeSeenTargets struct {
 	sync.RWMutex
-	data map[string]bool
+	data          map[string]bool
+	maxSize       int
+	limitExceeded atomic.Bool
+	clearingTimer *time.Ticker
 }
 
-func newSafeSeenTargets() SafeAnalyticsCache[string, bool] {
-	return &safeSeenTargets{
-		data: make(map[string]bool),
+// Implements SafeSeenTargetsCache
+func newSafeSeenTargets(maxSize int, clearingInterval time.Duration) SafeSeenTargetsCache[string, bool] {
+	st := &safeSeenTargets{
+		data:    make(map[string]bool),
+		maxSize: maxSize,
 	}
+
+	if clearingInterval > 0 {
+		st.clearingTimer = time.NewTicker(clearingInterval)
+
+		// Start a goroutine to clear the cache at a set interval
+		go func() {
+			for range st.clearingTimer.C {
+				st.clear()
+			}
+		}()
+	}
+
+	return st
 }
 
-func (s *safeSeenTargets) set(key string, seen bool) {
+func (s *safeSeenTargets) setWithLimit(key string, seen bool) {
+	if s.limitExceeded.Load() {
+		return
+	}
+
 	s.Lock()
 	defer s.Unlock()
+
+	if len(s.data) >= s.maxSize {
+		s.limitExceeded.Store(true)
+	}
+
 	s.data[key] = seen
 }
 
+// The regular set method just calls SetWithLimit
+func (s *safeSeenTargets) set(key string, seen bool) {
+	s.setWithLimit(key, seen)
+}
+
 func (s *safeSeenTargets) get(key string) (bool, bool) {
 	s.RLock()
 	defer s.RUnlock()
@@ -44,6 +78,7 @@ func (s *safeSeenTargets) clear() {
 	s.Lock()
 	defer s.Unlock()
 	s.data = make(map[string]bool)
+	s.limitExceeded.Store(false)
 }
 
 func (s *safeSeenTargets) iterate(f func(string, bool)) {
@@ -53,3 +88,7 @@ func (s *safeSeenTargets) iterate(f func(string, bool)) {
 		f(key, value)
 	}
 }
+
+func (s *safeSeenTargets) isLimitExceeded() bool {
+	return s.limitExceeded.Load()
+}

client/client.go

@@ -79,7 +79,7 @@ func NewCfClient(sdkKey string, options ...ConfigOption) (*CfClient, error) {
 		opt(config)
 	}
 
-	analyticsService := analyticsservice.NewAnalyticsService(time.Minute, config.Logger)
+	analyticsService := analyticsservice.NewAnalyticsService(time.Minute, config.Logger, config.seenTargetsMaxSize, config.seenTargetsClearInterval)
 
 	client := &CfClient{
 		sdkKey:                 sdkKey,

client/config.go

@@ -18,26 +18,28 @@ import (
 )
 
 type config struct {
-	url                    string
-	eventsURL              string
-	pullInterval           uint // in seconds
-	Cache                  cache.Cache
-	Store                  storage.Storage
-	Logger                 logger.Logger
-	httpClient             *http.Client
-	authHttpClient         *http.Client
-	enableStream           bool
-	enableStore            bool
-	target                 evaluation.Target
-	eventStreamListener    stream.EventStreamListener
-	enableAnalytics        bool
-	proxyMode              bool
-	waitForInitialized     bool
-	maxAuthRetries         int
-	authRetryStrategy      *backoff.ExponentialBackOff
-	streamingRetryStrategy *backoff.ExponentialBackOff
-	sleeper                types.Sleeper
-	apiConfig              *apiConfiguration
+	url                      string
+	eventsURL                string
+	pullInterval             uint // in seconds
+	Cache                    cache.Cache
+	Store                    storage.Storage
+	Logger                   logger.Logger
+	httpClient               *http.Client
+	authHttpClient           *http.Client
+	enableStream             bool
+	enableStore              bool
+	target                   evaluation.Target
+	eventStreamListener      stream.EventStreamListener
+	enableAnalytics          bool
+	proxyMode                bool
+	waitForInitialized       bool
+	maxAuthRetries           int
+	authRetryStrategy        *backoff.ExponentialBackOff
+	streamingRetryStrategy   *backoff.ExponentialBackOff
+	sleeper                  types.Sleeper
+	apiConfig                *apiConfiguration
+	seenTargetsMaxSize       int
+	seenTargetsClearInterval time.Duration
 }
 
 type apiConfiguration struct {
@@ -87,24 +89,25 @@ func newDefaultConfig(log logger.Logger) *config {
 	}
 
 	return &config{
-		url:             "https://config.ff.harness.io/api/1.0",
-		eventsURL:       "https://events.ff.harness.io/api/1.0",
-		pullInterval:    60,
-		Cache:           defaultCache,
-		Store:           defaultStore,
-		Logger:          log,
-		authHttpClient:  authHttpClient,
-		httpClient:      requestHttpClient.StandardClient(),
-		enableStream:    true,
-		enableStore:     true,
-		enableAnalytics: true,
-		proxyMode:       false,
-		// Indicate that we should retry forever by default
-		maxAuthRetries:         -1,
-		authRetryStrategy:      getDefaultExpBackoff(),
-		streamingRetryStrategy: getDefaultExpBackoff(),
-		sleeper:                &types.RealClock{},
-		apiConfig:              apiConfig,
+		url:                      "https://config.ff.harness.io/api/1.0",
+		eventsURL:                "https://events.ff.harness.io/api/1.0",
+		pullInterval:             60,
+		Cache:                    defaultCache,
+		Store:                    defaultStore,
+		Logger:                   log,
+		authHttpClient:           authHttpClient,
+		httpClient:               requestHttpClient.StandardClient(),
+		enableStream:             true,
+		enableStore:              true,
+		enableAnalytics:          true,
+		proxyMode:                false,
+		maxAuthRetries:           -1, // Indicate that we should retry forever by default
+		authRetryStrategy:        getDefaultExpBackoff(),
+		streamingRetryStrategy:   getDefaultExpBackoff(),
+		sleeper:                  &types.RealClock{},
+		apiConfig:                apiConfig,
+		seenTargetsMaxSize:       500000,
+		seenTargetsClearInterval: 24 * time.Hour,
 	}
 }
 

client/options.go

@@ -1,14 +1,16 @@
 package client
 
 import (
+	"net/http"
+	"time"
+
 	"github.com/cenkalti/backoff/v4"
 	"github.com/harness/ff-golang-server-sdk/cache"
 	"github.com/harness/ff-golang-server-sdk/evaluation"
 	"github.com/harness/ff-golang-server-sdk/logger"
 	"github.com/harness/ff-golang-server-sdk/storage"
 	"github.com/harness/ff-golang-server-sdk/stream"
 	"github.com/harness/ff-golang-server-sdk/types"
-	"net/http"
 )
 
 // ConfigOption is used as return value for advanced client configuration
@@ -142,3 +144,21 @@ func WithSleeper(sleeper types.Sleeper) ConfigOption {
 		config.sleeper = sleeper
 	}
 }
+
+// WithSeenTargetsMaxSize sets the maximum size for the seen targets map.
+// The SeenTargetsCache helps to reduce the size of the analytics payload that the SDK sends to the Feature Flags Service.
+// This method allows you to set the maximum number of unique targets that will be stored in the SeenTargets cache.
+// By default, the limit is set to 500,000 unique targets. You can increase this number if you need to handle more than
+// 500,000 targets, which will reduce the payload size but will also increase memory usage.
+func WithSeenTargetsMaxSize(maxSize int) ConfigOption {
+	return func(config *config) {
+		config.seenTargetsMaxSize = maxSize
+	}
+}
+
+// WithSeenTargetsClearInterval sets the clearing interval for the seen targets map.
+func WithSeenTargetsClearInterval(interval time.Duration) ConfigOption {
+	return func(config *config) {
+		config.seenTargetsClearInterval = interval
+	}
+}

evaluation/util.go

@@ -80,7 +80,7 @@ func isEnabled(target *Target, bucketBy string, percentage int) bool {
 		if value == "" {
 			return false
 		}
-		log.Warnf("%s BucketBy attribute not found in target attributes, falling back to 'identifier': missing=%s, using value=%s", sdk_codes.MissingBucketBy, bucketBy, value)
+		log.Debugf("%s BucketBy attribute not found in target attributes, falling back to 'identifier': missing=%s, using value=%s", sdk_codes.MissingBucketBy, bucketBy, value)
 		bucketBy = "identifier"
 	}