chore: rework metrics to use less locking on the hotpath

Author: sighphyre

metrics.go

@@ -10,6 +10,7 @@ import (
 	"net/url"
 	"runtime"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/Unleash/unleash-go-sdk/v5/internal/api"
@@ -87,20 +88,28 @@ type metric struct {
 	Enabled bool
 }
 
+type toggleCounters struct {
+	yes int64
+	no  int64
+
+	mu       sync.Mutex
+	variants map[string]int64
+}
+
 type metrics struct {
 	metricsChannels
-	options  metricsOptions
-	started  time.Time
-	bucketMu sync.Mutex
-	bucket   api.Bucket
-	ticker   *time.Ticker
-	close    chan struct{}
-	closed   chan struct{}
-	ctx      context.Context
-	cancel   func()
-	maxSkips float64
-	errors   float64
-	skips    float64
+	options         metricsOptions
+	started         time.Time
+	last_close_time time.Time
+	counters        sync.Map // map[string]*toggleCounters
+	ticker          *time.Ticker
+	close           chan struct{}
+	closed          chan struct{}
+	ctx             context.Context
+	cancel          func()
+	maxSkips        float64
+	errors          float64
+	skips           float64
 }
 
 func newMetrics(options metricsOptions, channels metricsChannels) *metrics {
@@ -113,6 +122,7 @@ func newMetrics(options metricsOptions, channels metricsChannels) *metrics {
 		maxSkips:        10,
 		errors:          0,
 		skips:           0,
+		last_close_time: time.Now(),
 	}
 	ctx, cancel := context.WithCancel(context.Background())
 	m.ctx = ctx
@@ -122,7 +132,6 @@ func newMetrics(options metricsOptions, channels metricsChannels) *metrics {
 		m.options.httpClient = http.DefaultClient
 	}
 
-	m.resetBucket()
 	if m.options.metricsInterval <= 0 {
 		m.options.disableMetrics = true
 	}
@@ -196,13 +205,70 @@ func (m *metrics) successfulPost() {
 func (m *metrics) decrementSkip() {
 	m.skips = math.Max(0, m.skips-1)
 }
+
+// This does not remove stale toggle names from the map. I don't think there's a safe, lock free way to do that
+// The consequence is that if the user archives a lot of toggles this internal representation will not lose those
+// toggles until the process is terminated. In practice, I don't believe this is a big problem, just means a
+// little bit more memory is held than necessary
+func (m *metrics) buildBucketAndReset(last_close_time time.Time) (api.Bucket, bool) {
+	bucket := api.Bucket{
+		Start:   last_close_time,
+		Toggles: make(map[string]api.ToggleCount),
+	}
+
+	m.counters.Range(func(key, value any) bool {
+		name := key.(string)
+		c := value.(*toggleCounters)
+
+		yes := atomic.SwapInt64(&c.yes, 0)
+		no := atomic.SwapInt64(&c.no, 0)
+
+		if yes == 0 && no == 0 {
+			c.mu.Lock()
+			emptyVariants := len(c.variants) == 0
+			c.mu.Unlock()
+			if emptyVariants {
+				return true
+			}
+		}
+
+		tc := api.ToggleCount{
+			Yes: int32(yes),
+			No:  int32(no),
+		}
+
+		// we can have a little locking, as a treat. Variants are likely a luke warm path at best
+		// until we have evidence that this is a hot path API, I'd like to keep this simple
+		// simple here means a local lock per toggle counter while we swap out the variants map
+		c.mu.Lock()
+		if len(c.variants) > 0 {
+			vars := make(map[string]int32, len(c.variants))
+			for vName, cnt := range c.variants {
+				vars[vName] = int32(cnt)
+			}
+			tc.Variants = vars
+
+			c.variants = make(map[string]int64)
+		}
+		c.mu.Unlock()
+
+		bucket.Toggles[name] = tc
+		return true
+	})
+
+	if len(bucket.Toggles) == 0 {
+		return api.Bucket{}, false
+	}
+
+	return bucket, true
+}
+
 func (m *metrics) sendMetrics() {
-	m.bucketMu.Lock()
-	bucket := m.resetBucket()
-	m.bucketMu.Unlock()
-	if bucket.IsEmpty() {
+	bucket, ok := m.buildBucketAndReset(m.last_close_time)
+	if !ok {
 		return
 	}
+	m.last_close_time = time.Now()
 	bucket.Stop = time.Now()
 	payload := MetricsData{
 		AppName:          m.options.appName,
@@ -233,23 +299,19 @@ func (m *metrics) sendMetrics() {
 		m.warn(fmt.Errorf("%s return %d", u.String(), resp.StatusCode))
 		// The post failed, re-add the metrics we attempted to send so
 		// they are included in the next post.
-		for name, tc := range bucket.Toggles {
-			m.add(name, true, tc.Yes)
-			m.add(name, false, tc.No)
-		}
+		m.reinsertBucket(bucket)
 
-		m.bucketMu.Lock()
 		// Set the start time of the current bucket to the one we
 		// attempted to send.
-		m.bucket.Start = bucket.Start
-		m.bucketMu.Unlock()
+		m.last_close_time = bucket.Start
+
 	} else {
 		m.successfulPost()
 		m.sent <- payload
 	}
 }
 
-func (m *metrics) doPost(url *url.URL, payload any) (*http.Response, error) {
+func (m *metrics) doPost(url *url.URL, payload interface{}) (*http.Response, error) {
 	var body bytes.Buffer
 	enc := json.NewEncoder(&body)
 	if err := enc.Encode(payload); err != nil {
@@ -274,24 +336,57 @@ func (m *metrics) doPost(url *url.URL, payload any) (*http.Response, error) {
 	return m.options.httpClient.Do(req)
 }
 
+func (m *metrics) getOrCreateCounter(name string) *toggleCounters {
+	c, ok := m.counters.Load(name)
+	if ok {
+		return c.(*toggleCounters)
+	}
+
+	nc := &toggleCounters{
+		variants: make(map[string]int64),
+	}
+	actual, _ := m.counters.LoadOrStore(name, nc)
+	return actual.(*toggleCounters)
+}
+
+func (m *metrics) reinsertBucket(bucket api.Bucket) {
+	for name, tc := range bucket.Toggles {
+		c := m.getOrCreateCounter(name)
+		if tc.Yes != 0 {
+			atomic.AddInt64(&c.yes, int64(tc.Yes))
+		}
+		if tc.No != 0 {
+			atomic.AddInt64(&c.no, int64(tc.No))
+		}
+
+		if len(tc.Variants) > 0 {
+			c := m.getOrCreateCounter(name)
+
+			c.mu.Lock()
+			if c.variants == nil {
+				c.variants = make(map[string]int64, len(tc.Variants))
+			}
+			for vName, cnt := range tc.Variants {
+				if cnt == 0 {
+					continue
+				}
+				c.variants[vName] += int64(cnt)
+			}
+			c.mu.Unlock()
+		}
+	}
+}
+
 func (m *metrics) add(name string, enabled bool, num int32) {
 	if m.options.disableMetrics || num == 0 {
 		return
 	}
-	m.bucketMu.Lock()
-	defer m.bucketMu.Unlock()
-	t, exists := m.bucket.Toggles[name]
-	if !exists {
-		t = api.ToggleCount{
-			Variants: map[string]int32{},
-		}
-	}
+	c := m.getOrCreateCounter(name)
 	if enabled {
-		t.Yes += num
+		atomic.AddInt64(&c.yes, int64(num))
 	} else {
-		t.No += num
+		atomic.AddInt64(&c.no, int64(num))
 	}
-	m.bucket.Toggles[name] = t
 }
 
 func (m *metrics) count(name string, enabled bool) {
@@ -310,29 +405,14 @@ func (m *metrics) countVariants(name string, enabled bool, variantName string) {
 	m.add(name, enabled, 1)
 	m.metricsChannels.count <- metric{Name: name, Enabled: enabled}
 
-	m.bucketMu.Lock()
-	defer m.bucketMu.Unlock()
-
-	t := m.bucket.Toggles[name]
-	if len(t.Variants) == 0 {
-		t.Variants = make(map[string]int32)
-	}
-
-	if _, ok := t.Variants[variantName]; !ok {
-		t.Variants[variantName] = 1
-	} else {
-		t.Variants[variantName] += 1
-	}
-	m.bucket.Toggles[name] = t
-}
+	c := m.getOrCreateCounter(name)
 
-func (m *metrics) resetBucket() api.Bucket {
-	prev := m.bucket
-	m.bucket = api.Bucket{
-		Start:   time.Now(),
-		Toggles: map[string]api.ToggleCount{},
+	c.mu.Lock()
+	if c.variants == nil {
+		c.variants = make(map[string]int64)
 	}
-	return prev
+	c.variants[variantName]++
+	c.mu.Unlock()
 }
 
 func (m *metrics) getClientData() ClientData {

metrics_test.go

@@ -86,7 +86,9 @@ func TestMetrics_VariantsCountToggles(t *testing.T) {
 	client.WaitForReady()
 	client.GetVariant("foo")
 
-	assert.EqualValues(client.metrics.bucket.Toggles["foo"].No, 1)
+	registered_metric, ok := client.metrics.counters.Load("foo")
+	assert.True(ok, "should have a count for 'foo'")
+	assert.EqualValues(registered_metric.(*toggleCounters).no, 1)
 	client.Close()
 
 	assert.Nil(err, "client should not return an error")
@@ -335,8 +337,13 @@ func TestMetrics_ShouldNotCountMetricsForParentToggles(t *testing.T) {
 	client.WaitForReady()
 	client.IsEnabled("child")
 
-	assert.EqualValues(client.metrics.bucket.Toggles["child"].Yes, 1)
-	assert.EqualValues(client.metrics.bucket.Toggles["parent"].Yes, 0)
+	child_metric, ok := client.metrics.counters.Load("child")
+	assert.True(ok, "should have a count for 'child'")
+	assert.EqualValues(child_metric.(*toggleCounters).yes, 1)
+
+	_, ok = client.metrics.counters.Load("parent")
+	assert.False(ok, "should not have a count for parent'")
+
 	err = client.Close()
 
 	assert.Nil(err, "client should not return an error")
@@ -592,3 +599,36 @@ func TestMetrics_metricsData_includes_new_metadata(t *testing.T) {
 
 	st.Expect(t, gock.IsDone(), true)
 }
+
+func TestReinsertingBucketsAlsoRestoresVariants(t *testing.T) {
+	metrics_handler := &metrics{
+		metricsChannels: metricsChannels{
+			count: make(chan metric, 100),
+		},
+	}
+
+	metrics_handler.countVariants("some-feature", true, "some-variant")
+	registered_metric, ok := metrics_handler.counters.Load("some-feature")
+
+	if !ok {
+		t.Fatal("should have a count for 'some-feature'")
+	}
+
+	assert.EqualValues(t, 1, registered_metric.(*toggleCounters).variants["some-variant"])
+
+	retrieved_bucket, ok := metrics_handler.buildBucketAndReset(time.Now())
+	if !ok {
+		t.Fatal("Missing bucket for 'some-feature'")
+	}
+
+	assert.EqualValues(t, 1, retrieved_bucket.Toggles["some-feature"].Variants["some-variant"])
+
+	metrics_handler.reinsertBucket(retrieved_bucket)
+
+	registered_metric, ok = metrics_handler.counters.Load("some-feature")
+	if !ok {
+		t.Fatal("should have a count for 'some-feature'")
+	}
+
+	assert.EqualValues(t, 1, registered_metric.(*toggleCounters).variants["some-variant"])
+}