bugfix: mergeSeries should normalize its inputs. fix #1673

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"
@@ -624,7 +626,7 @@ func (s *Server) getSeriesCachedStore(ctx *requestContext, ss *models.StorageSta
 // 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, data map[expr.Req][]models.Series) []models.Series {
 	type segment struct {
 		target  string
 		query   string
@@ -656,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(data, 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))
 	}

api/graphite.go

@@ -779,7 +779,13 @@ func (s *Server) executePlan(ctx context.Context, orgId uint32, plan expr.Plan)
 	meta.RenderStats.GetTargetsDuration = b.Sub(a)
 	meta.StorageStats.Trace(span)
 
-	out = mergeSeries(out)
+	// this map will contain all series that we will feed into the processing chain or generated therein:
+	// * fetched series, grouped by their expr.Req, such that expr.FuncGet can find the data it needs and feed it into subsequent expr.GraphiteFunc functions
+	// * additional series generated while handling the request (e.g. function processing, normalization), keyed by an empty expr.Req (such that can't be mistakenly picked up by FuncGet)
+	// all of these series will need to be returned to the pool once we're done with all processing and have generated our response body, which will happen in plan.Clean()
+	data := make(map[expr.Req][]models.Series)
+
+	out = mergeSeries(out, data)
 
 	if len(metaTagEnrichmentData) > 0 {
 		for i := range out {
@@ -792,7 +798,6 @@ func (s *Server) executePlan(ctx context.Context, orgId uint32, plan expr.Plan)
 	// instead of waiting for all data to come in and then start processing everything, we could consider starting processing earlier, at the risk of doing needless work
 	// if we need to cancel the request due to a fetch error
 
-	data := make(map[expr.Req][]models.Series)
 	for _, serie := range out {
 		q := expr.NewReqFromSerie(serie)
 		data[q] = append(data[q], serie)