Merge pull request #1425 from grafana/flush-boundary-in-both-cases

when adding new point to aggregate, always see if we can flush

Author: Dieterbe

mdata/aggmetric_test.go

@@ -407,6 +407,7 @@ func TestGetAggregated(t *testing.T) {
 		{Val: 21 + 22, Ts: 25},
 		{Val: 30, Ts: 30},
 		{Val: 31 + 32, Ts: 35},
+		{Val: 40, Ts: 40},
 	}
 	assertPointsEqual(t, got, expected)
 }
@@ -454,6 +455,7 @@ func TestGetAggregatedIngestFrom(t *testing.T) {
 	expected := []schema.Point{
 		{Val: 26 + 30, Ts: 30},
 		{Val: 31 + 32, Ts: 35},
+		{Val: 40, Ts: 40},
 	}
 	assertPointsEqual(t, got, expected)
 }

mdata/aggregator.go

@@ -100,19 +100,26 @@ func (agg *Aggregator) flush() {
 func (agg *Aggregator) Add(ts uint32, val float64) {
 	boundary := AggBoundary(ts, agg.span)
 
-	if boundary == agg.currentBoundary {
-		agg.agg.Add(val)
-		if ts == boundary {
-			agg.flush()
-		}
+	if boundary < agg.currentBoundary {
+		// ignore the point it was for a previous bucket. we can't process it
+		return
 	} else if boundary > agg.currentBoundary {
-		// store current totals as a new point in their series
+		// point is for a more recent bucket
+		// store current aggregates as a new point in their series and start the new bucket
 		// if the cnt is still 0, the numbers are invalid, not to be flushed and we can simply reuse the aggregation
 		if agg.agg.Cnt != 0 {
 			agg.flush()
 		}
 		agg.currentBoundary = boundary
-		agg.agg.Add(val)
+	}
+	agg.agg.Add(val)
+
+	// if the ts of the point is a boundary, it means no more point can possibly come in for the same aggregation.
+	// e.g. if aggspan is 10s and we're adding a point with timestamp 12:34:10, then any subsequent point will go
+	// in the bucket for 12:34:20 or later.
+	// so it is time to flush the result
+	if ts == boundary {
+		agg.flush()
 	}
 }