Merge pull request #1674 from grafana/mergeSeries-runtime-normalize

fix: mergeSeries() should normalize its inputs + normalize() bugfix + cleanups

Author: Dieterbe

api/dataprocessor.go

@@ -9,6 +9,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/grafana/metrictank/expr"
+
 	"github.com/grafana/metrictank/api/models"
 	"github.com/grafana/metrictank/consolidation"
 	"github.com/grafana/metrictank/mdata"
@@ -619,11 +621,12 @@ func (s *Server) getSeriesCachedStore(ctx *requestContext, ss *models.StorageSta
 	return iters, nil
 }
 
-// check for duplicate series names for the same query target. If found merge the results.
+// mergeSeries merges series together if applicable. It does this by categorizing
+// series into groups based on their target, query, consolidator etc. If they collide, they get merged.
 // each first uniquely-identified series's backing datapoints slice is reused
 // any subsequent non-uniquely-identified series is merged into the former and has its
 // datapoints slice returned to the pool. input series must be canonical
-func mergeSeries(in []models.Series) []models.Series {
+func mergeSeries(in []models.Series, dataMap expr.DataMap) []models.Series {
 	type segment struct {
 		target  string
 		query   string
@@ -655,6 +658,7 @@ func mergeSeries(in []models.Series) []models.Series {
 			// we use the first series in the list as our result.  We check over every
 			// point and if it is null, we then check the other series for a non null
 			// value to use instead.
+			series = expr.Normalize(dataMap, series)
 			log.Debugf("DP mergeSeries: %s has multiple series.", series[0].Target)
 			for i := range series[0].Datapoints {
 				for j := 0; j < len(series); j++ {

api/dataprocessor_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/grafana/metrictank/cluster"
 	"github.com/grafana/metrictank/conf"
 	"github.com/grafana/metrictank/consolidation"
+	"github.com/grafana/metrictank/expr"
 	"github.com/grafana/metrictank/mdata"
 	"github.com/grafana/metrictank/mdata/cache"
 	"github.com/grafana/metrictank/mdata/cache/accnt"
@@ -628,7 +629,7 @@ func TestMergeSeries(t *testing.T) {
 		Interval: 10,
 	})
 
-	merged := mergeSeries(out)
+	merged := mergeSeries(out, expr.NewDataMap())
 	if len(merged) != 5 {
 		t.Errorf("Expected data to be merged down to 5 series. got %d instead", len(merged))
 	}