[DO NOT LAND] Process only changed metrics

scope.go

@@ -102,6 +102,9 @@
 	done        chan struct{}
 	wg          sync.WaitGroup
 	root        bool
+
+	counterChangeNotifyCh chan *counter
+	gaugeChangeNotifyCh   chan *gauge
 }
 
 // ScopeOptions is a set of options to construct a scope.
@@ -166,23 +169,25 @@
 	}
 
 	s := &scope{
-		baseReporter:    baseReporter,
-		bucketCache:     newBucketCache(),
-		cachedReporter:  opts.CachedReporter,
-		counters:        make(map[string]*counter),
-		countersSlice:   make([]*counter, 0, _defaultInitialSliceSize),
-		defaultBuckets:  opts.DefaultBuckets,
-		done:            make(chan struct{}),
-		gauges:          make(map[string]*gauge),
-		gaugesSlice:     make([]*gauge, 0, _defaultInitialSliceSize),
-		histograms:      make(map[string]*histogram),
-		histogramsSlice: make([]*histogram, 0, _defaultInitialSliceSize),
-		prefix:          sanitizer.Name(opts.Prefix),
-		reporter:        opts.Reporter,
-		sanitizer:       sanitizer,
-		separator:       sanitizer.Name(opts.Separator),
-		timers:          make(map[string]*timer),
-		root:            true,
+		baseReporter:          baseReporter,
+		bucketCache:           newBucketCache(),
+		cachedReporter:        opts.CachedReporter,
+		counters:              make(map[string]*counter),
+		countersSlice:         make([]*counter, 0, _defaultInitialSliceSize),
+		defaultBuckets:        opts.DefaultBuckets,
+		done:                  make(chan struct{}),
+		gauges:                make(map[string]*gauge),
+		gaugesSlice:           make([]*gauge, 0, _defaultInitialSliceSize),
+		histograms:            make(map[string]*histogram),
+		histogramsSlice:       make([]*histogram, 0, _defaultInitialSliceSize),
+		prefix:                sanitizer.Name(opts.Prefix),
+		reporter:              opts.Reporter,
+		sanitizer:             sanitizer,
+		separator:             sanitizer.Name(opts.Separator),
+		timers:                make(map[string]*timer),
+		root:                  true,
+		counterChangeNotifyCh: make(chan *counter, 1024),
+		gaugeChangeNotifyCh:   make(chan *gauge, 1024),
 	}
 
 	// NB(r): Take a copy of the tags on creation
@@ -196,7 +201,7 @@
 		s.wg.Add(1)
 		go func() {
 			defer s.wg.Done()
-			s.reportLoop(interval)
+			s.processLoop(interval)
 		}()
 	}
 
@@ -249,7 +254,7 @@
 }
 
 // reportLoop is used by the root scope for periodic reporting
 func (s *scope) reportLoop(interval time.Duration) {
 	ticker := time.NewTicker(interval)
 	defer ticker.Stop()
 
@@ -281,6 +286,39 @@
 	}
 }
 
+func (s *scope) processLoop(interval time.Duration) {
+	ticker := time.NewTicker(interval)
+	counters := make([]*counter, 0, _defaultInitialSliceSize)
+	gauges := make([]*gauge, 0, _defaultInitialSliceSize)
+
+	defer ticker.Stop()
+	for {
+		select {
+		case c := <-s.counterChangeNotifyCh:
+			counters = append(counters, c)
+		case g := <-s.gaugeChangeNotifyCh:
+			gauges = append(gauges, g)
+		case <-ticker.C:
+			s.reportChanges(counters, gauges)
+			s.cachedReporter.Flush()
+			// Reset the changed counters and gauges
+			counters = counters[:0]
+			gauges = gauges[:0]
+		default:
+			return
+		}
+	}
+}
+
+func (s *scope) reportChanges(counters []*counter, gauges []*gauge) {
+	for _, c := range counters {
+		c.cachedReport()
+	}
+	for _, g := range gauges {
+		g.cachedReport()
+	}
+}
+
 func (s *scope) Counter(name string) Counter {
 	name = s.sanitizer.Name(name)
 	if c, ok := s.counter(name); ok {
@@ -295,14 +333,18 @@
 	}
 
 	var cachedCounter CachedCount
+	var changeNotifyFn func(c *counter)
 	if s.cachedReporter != nil {
 		cachedCounter = s.cachedReporter.AllocateCounter(
 			s.fullyQualifiedName(name),
 			s.tags,
 		)
+		changeNotifyFn = func(c *counter) {
+			s.counterChangeNotifyCh <- c
+		}
 	}
 
-	c := newCounter(cachedCounter)
+	c := newCounter(cachedCounter, changeNotifyFn)
 	s.counters[name] = c
 	s.countersSlice = append(s.countersSlice, c)
 
@@ -331,13 +373,17 @@
 	}
 
 	var cachedGauge CachedGauge
+	var changeNotifyFn func(g *gauge)
 	if s.cachedReporter != nil {
 		cachedGauge = s.cachedReporter.AllocateGauge(
 			s.fullyQualifiedName(name), s.tags,
 		)
+		changeNotifyFn = func(g *gauge) {
+			s.gaugeChangeNotifyCh <- g
+		}
 	}
 
-	g := newGauge(cachedGauge)
+	g := newGauge(cachedGauge, changeNotifyFn)
 	s.gauges[name] = g
 	s.gaugesSlice = append(s.gaugesSlice, g)
 

scope_registry.go

@@ -196,6 +196,9 @@ func (r *scopeRegistry) Subscope(parent *scope, prefix string, tags map[string]s
 		timers:          make(map[string]*timer),
 		bucketCache:     parent.bucketCache,
 		done:            make(chan struct{}),
+
+		counterChangeNotifyCh: parent.counterChangeNotifyCh,
+		gaugeChangeNotifyCh:   parent.gaugeChangeNotifyCh,
 	}
 	subscopeBucket.s[key] = subscope
 	if _, ok := r.lockedLookup(subscopeBucket, preSanitizeKey); !ok {

stats.go

@@ -60,17 +60,30 @@ func (c *capabilities) Tagging() bool {
 }
 
 type counter struct {
-	prev        int64
-	curr        int64
-	cachedCount CachedCount
-}
-
-func newCounter(cachedCount CachedCount) *counter {
-	return &counter{cachedCount: cachedCount}
+	prev           int64
+	curr           int64
+	updated        uint64
+	cachedCount    CachedCount
+	changeNotifyFn func(c *counter)
+}
+
+func newCounter(
+	cachedCount CachedCount,
+	changeNotifyFn func(c *counter),
+) *counter {
+	return &counter{
+		cachedCount:    cachedCount,
+		changeNotifyFn: changeNotifyFn,
+	}
 }
 
 func (c *counter) Inc(v int64) {
 	atomic.AddInt64(&c.curr, v)
+	if c.changeNotifyFn != nil {
+		if atomic.SwapUint64(&c.updated, 1) == 0 {
+			c.changeNotifyFn(c)
+		}
+	}
 }
 
 func (c *counter) value() int64 {
@@ -99,26 +112,39 @@ func (c *counter) cachedReport() {
 		return
 	}
 
-	c.cachedCount.ReportCount(delta)
+	if atomic.SwapUint64(&c.updated, 0) == 1 {
+		c.cachedCount.ReportCount(delta)
+	}
 }
 
 func (c *counter) snapshot() int64 {
 	return atomic.LoadInt64(&c.curr) - atomic.LoadInt64(&c.prev)
 }
 
 type gauge struct {
-	updated     uint64
-	curr        uint64
-	cachedGauge CachedGauge
-}
-
-func newGauge(cachedGauge CachedGauge) *gauge {
-	return &gauge{cachedGauge: cachedGauge}
+	updated        uint64
+	curr           uint64
+	cachedGauge    CachedGauge
+	changeNotifyFn func(g *gauge)
+}
+
+func newGauge(
+	cachedGauge CachedGauge,
+	changeNotifyFn func(g *gauge),
+) *gauge {
+	return &gauge{
+		cachedGauge:    cachedGauge,
+		changeNotifyFn: changeNotifyFn,
+	}
 }
 
 func (g *gauge) Update(v float64) {
 	atomic.StoreUint64(&g.curr, math.Float64bits(v))
-	atomic.StoreUint64(&g.updated, 1)
+	if atomic.SwapUint64(&g.updated, 1) == 0 {
+		if g.changeNotifyFn != nil {
+			g.changeNotifyFn(g)
+		}
+	}
 }
 
 func (g *gauge) value() float64 {
@@ -297,7 +323,7 @@ func newHistogram(
 	}
 
 	for i := range h.samples {
-		h.samples[i].counter = newCounter(nil)
+		h.samples[i].counter = newCounter(nil, nil)
 
 		if cachedHistogram != nil {
 			switch htype {

stats_test.go

@@ -84,7 +84,7 @@ func (r *statsTestReporter) Capabilities() Capabilities {
 func (r *statsTestReporter) Flush() {}
 
 func TestCounter(t *testing.T) {
-	counter := newCounter(nil)
+	counter := newCounter(nil, nil)
 	r := newStatsTestReporter()
 
 	counter.Inc(1)
@@ -101,7 +101,7 @@ func TestCounter(t *testing.T) {
 }
 
 func TestGauge(t *testing.T) {
-	gauge := newGauge(nil)
+	gauge := newGauge(nil, nil)
 	r := newStatsTestReporter()
 
 	gauge.Update(42)