aggregators now need to be able to normalize at runtime

Author: Dieterbe

expr/func_aggregate.go

@@ -48,6 +48,7 @@ func (s *FuncAggregate) Exec(cache map[Req][]models.Series) ([]models.Series, er
 		return series, nil
 	}
 	out := pointSlicePool.Get().([]schema.Point)
+	series = normalize(series)
 	s.agg.function(series, &out)
 
 	// The tags for the aggregated series is only the tags that are

expr/func_groupbytags.go

@@ -130,7 +130,7 @@ func (s *FuncGroupByTags) Exec(cache map[Req][]models.Series) ([]models.Series,
 		newSeries.SetTags()
 
 		newSeries.Datapoints = pointSlicePool.Get().([]schema.Point)
-
+		group.s = normalize(group.s)
 		aggFunc(group.s, &newSeries.Datapoints)
 		cache[Req{}] = append(cache[Req{}], newSeries)
 

expr/normalize.go

@@ -0,0 +1,27 @@
+package expr
+
+import (
+	"github.com/grafana/metrictank/api/models"
+	"github.com/grafana/metrictank/consolidation"
+	"github.com/grafana/metrictank/schema"
+	"github.com/grafana/metrictank/util"
+)
+
+// normalize normalizes series to the same common LCM interval - if they don't already have the same interval
+func normalize(in []models.Series) []models.Series {
+	var intervals []uint32
+	for _, s := range in {
+		intervals = append(intervals, s.Interval)
+	}
+	lcm := util.Lcm(intervals)
+	for i, s := range in {
+		if s.Interval != lcm {
+			// we need to copy the datapoints first because the consolidater will reuse the input slice
+			datapoints := pointSlicePool.Get().([]schema.Point)
+			datapoints = append(datapoints, s.Datapoints...)
+			consolidation.Consolidate(datapoints, lcm/s.Interval, s.Consolidator) // TODO: not sure if we should use s.Consolidator here
+			in[i].Datapoints = datapoints
+		}
+	}
+	return in
+}