Better implementation of Aggregator ingestFrom. Fixes failing test.

Author: fkaleo

mdata/aggregator.go

@@ -27,6 +27,7 @@ type Aggregator struct {
 	sumMetric       *AggMetric
 	cntMetric       *AggMetric
 	lstMetric       *AggMetric
+	ingestFrom      uint32
 }
 
 func NewAggregator(store Store, cachePusher cache.CachePusher, key schema.AMKey, ret conf.Retention, agg conf.Aggregation, dropFirstChunk bool, ingestFrom int64) *Aggregator {
@@ -38,36 +39,39 @@ func NewAggregator(store Store, cachePusher cache.CachePusher, key schema.AMKey,
 		span: span,
 		agg:  NewAggregation(),
 	}
+	if ingestFrom > 0 {
+		aggregator.ingestFrom = AggBoundary(uint32(ingestFrom), ret.ChunkSpan) - span
+	}
 	for _, agg := range agg.AggregationMethod {
 		switch agg {
 		case conf.Avg:
 			if aggregator.sumMetric == nil {
 				key.Archive = schema.NewArchive(schema.Sum, span)
-				aggregator.sumMetric = NewAggMetric(store, cachePusher, key, conf.Retentions{ret}, 0, span, nil, dropFirstChunk, ingestFrom)
+				aggregator.sumMetric = NewAggMetric(store, cachePusher, key, conf.Retentions{ret}, 0, span, nil, dropFirstChunk, 0)
 			}
 			if aggregator.cntMetric == nil {
 				key.Archive = schema.NewArchive(schema.Cnt, span)
-				aggregator.cntMetric = NewAggMetric(store, cachePusher, key, conf.Retentions{ret}, 0, span, nil, dropFirstChunk, ingestFrom)
+				aggregator.cntMetric = NewAggMetric(store, cachePusher, key, conf.Retentions{ret}, 0, span, nil, dropFirstChunk, 0)
 			}
 		case conf.Sum:
 			if aggregator.sumMetric == nil {
 				key.Archive = schema.NewArchive(schema.Sum, span)
-				aggregator.sumMetric = NewAggMetric(store, cachePusher, key, conf.Retentions{ret}, 0, span, nil, dropFirstChunk, ingestFrom)
+				aggregator.sumMetric = NewAggMetric(store, cachePusher, key, conf.Retentions{ret}, 0, span, nil, dropFirstChunk, 0)
 			}
 		case conf.Lst:
 			if aggregator.lstMetric == nil {
 				key.Archive = schema.NewArchive(schema.Lst, span)
-				aggregator.lstMetric = NewAggMetric(store, cachePusher, key, conf.Retentions{ret}, 0, span, nil, dropFirstChunk, ingestFrom)
+				aggregator.lstMetric = NewAggMetric(store, cachePusher, key, conf.Retentions{ret}, 0, span, nil, dropFirstChunk, 0)
 			}
 		case conf.Max:
 			if aggregator.maxMetric == nil {
 				key.Archive = schema.NewArchive(schema.Max, span)
-				aggregator.maxMetric = NewAggMetric(store, cachePusher, key, conf.Retentions{ret}, 0, span, nil, dropFirstChunk, ingestFrom)
+				aggregator.maxMetric = NewAggMetric(store, cachePusher, key, conf.Retentions{ret}, 0, span, nil, dropFirstChunk, 0)
 			}
 		case conf.Min:
 			if aggregator.minMetric == nil {
 				key.Archive = schema.NewArchive(schema.Min, span)
-				aggregator.minMetric = NewAggMetric(store, cachePusher, key, conf.Retentions{ret}, 0, span, nil, dropFirstChunk, ingestFrom)
+				aggregator.minMetric = NewAggMetric(store, cachePusher, key, conf.Retentions{ret}, 0, span, nil, dropFirstChunk, 0)
 			}
 		}
 	}
@@ -98,6 +102,9 @@ func (agg *Aggregator) flush() {
 // Add adds the point to the in-progress aggregation, and flushes it if we reached the boundary
 // points going back in time are accepted, unless they go into a previous bucket, in which case they are ignored
 func (agg *Aggregator) Add(ts uint32, val float64) {
+	if ts <= agg.ingestFrom {
+		return
+	}
 	boundary := AggBoundary(ts, agg.span)
 
 	if boundary == agg.currentBoundary {