Fix race when checking for dirty aggregations

src/aggregator/aggregation/common.go

@@ -44,7 +44,9 @@ func isExpensive(aggTypes aggregation.Types) bool {
 	return false
 }
 
-func maybeReplaceAnnotation(currentAnnotation, newAnnotation []byte) []byte {
+// MaybeReplaceAnnotation replaces the current annotation with the new annotation, returning the updated ref for
+// current annotation.
+func MaybeReplaceAnnotation(currentAnnotation, newAnnotation []byte) []byte {
 	if len(newAnnotation) == 0 {
 		return currentAnnotation
 	}

src/aggregator/aggregation/counter.go

@@ -74,7 +74,7 @@ func (c *Counter) Update(timestamp time.Time, value int64, annotation []byte) {
 		c.sumSq += value * value
 	}
 
-	c.annotation = maybeReplaceAnnotation(c.annotation, annotation)
+	c.annotation = MaybeReplaceAnnotation(c.annotation, annotation)
 }
 
 // LastAt returns the time of the last value received.

src/aggregator/aggregation/gauge.go

@@ -52,7 +52,7 @@ func NewGauge(opts Options) Gauge {
 
 // Update updates the gauge value.
 func (g *Gauge) Update(timestamp time.Time, value float64, annotation []byte) {
-	g.annotation = maybeReplaceAnnotation(g.annotation, annotation)
+	g.annotation = MaybeReplaceAnnotation(g.annotation, annotation)
 	g.updateTotals(timestamp, value)
 	// min/max cannot be updated by an update to a value.
 	if math.IsNaN(g.max) || g.max < value {

src/aggregator/aggregation/timer.go

@@ -72,7 +72,7 @@ func (t *Timer) AddBatch(timestamp time.Time, values []float64, annotation []byt
 
 	t.stream.AddBatch(values)
 
-	t.annotation = maybeReplaceAnnotation(t.annotation, annotation)
+	t.annotation = MaybeReplaceAnnotation(t.annotation, annotation)
 }
 
 func (t *Timer) recordLastAt(timestamp time.Time) {