Merge pull request #951 from grafana/refactor-alignrequests

remove excessive request alignment, add MDP optimization and pre-normalisation

Author: robert-milan

api/config.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/grafana/globalconf"
+	"github.com/grafana/metrictank/expr"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -28,6 +29,7 @@ var (
 	getTargetsConcurrency int
 	tagdbDefaultLimit     uint
 	speculationThreshold  float64
+	optimizations         expr.Optimizations
 
 	graphiteProxy *httputil.ReverseProxy
 	timeZone      *time.Location
@@ -49,6 +51,8 @@ func ConfigSetup() {
 	apiCfg.IntVar(&getTargetsConcurrency, "get-targets-concurrency", 20, "maximum number of concurrent threads for fetching data on the local node. Each thread handles a single series.")
 	apiCfg.UintVar(&tagdbDefaultLimit, "tagdb-default-limit", 100, "default limit for tagdb query results, can be overridden with query parameter \"limit\"")
 	apiCfg.Float64Var(&speculationThreshold, "speculation-threshold", 1, "ratio of peer responses after which speculation is used. Set to 1 to disable.")
+	apiCfg.BoolVar(&optimizations.PreNormalization, "pre-normalization", true, "enable pre-normalization optimization")
+	apiCfg.BoolVar(&optimizations.MDP, "mdp-optimization", false, "enable MaxDataPoints optimization (experimental)")
 	globalconf.Register("http", apiCfg, flag.ExitOnError)
 }
 

api/dataprocessor.go

@@ -329,7 +329,7 @@ LOOP:
 
 }
 
-// getTarget returns the series for the request in canonical form.
+// getTarget returns the series for the request in canonical form with respect to their OutInterval
 // as ConsolidateContext just processes what it's been given (not "stable" or bucket-aligned to the output interval)
 // we simply make sure to pass it the right input such that the output is canonical.
 func (s *Server) getTarget(ctx context.Context, ss *models.StorageStats, req models.Req) (out models.Series, err error) {
@@ -352,6 +352,8 @@ func (s *Server) getTarget(ctx context.Context, ss *models.StorageStats, req mod
 		QueryTo:      req.To,
 		QueryCons:    req.ConsReq,
 		Consolidator: req.Consolidator,
+		QueryMDP:     req.MaxPoints,
+		QueryPNGroup: req.PNGroup,
 		Meta: []models.SeriesMetaProperties{
 			{
 				// note that for simplicity, we pretend that a read of rollup avg data is a read of 1 "avg series"
@@ -405,7 +407,7 @@ func logLoad(typ string, key schema.AMKey, from, to uint32) {
 	log.Debugf("DP load from %-6s %20s %d - %d (%s - %s) span:%ds", typ, key, from, to, util.TS(from), util.TS(to), to-from-1)
 }
 
-// getSeriesFixed fetches the series and returns it in quantized, pre-canonical form.
+// getSeriesFixed fetches the series and returns it in quantized, pre-canonical form with respect to their OutInterval
 // TODO: we can probably forego Fix if archive > 0, because only raw chunks are not quantized yet.
 // the requested consolidator is the one that will be used for selecting the archive to read from
 func (s *Server) getSeriesFixed(ctx context.Context, ss *models.StorageStats, req models.Req, consolidator consolidation.Consolidator) ([]schema.Point, error) {
@@ -617,17 +619,19 @@ func (s *Server) getSeriesCachedStore(ctx *requestContext, ss *models.StorageSta
 	return iters, nil
 }
 
-// check for duplicate series names for the same query. If found merge the results.
+// check for duplicate series names for the same query target. If found merge the results.
 // 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 {
 	type segment struct {
-		target string
-		query  string
-		from   uint32
-		to     uint32
-		con    consolidation.Consolidator
+		target  string
+		query   string
+		from    uint32
+		to      uint32
+		con     consolidation.Consolidator
+		mdp     uint32
+		pngroup models.PNGroup
 	}
 	seriesByTarget := make(map[segment][]models.Series)
 	for _, series := range in {
@@ -637,6 +641,8 @@ func mergeSeries(in []models.Series) []models.Series {
 			series.QueryFrom,
 			series.QueryTo,
 			series.Consolidator,
+			series.QueryMDP,
+			series.QueryPNGroup,
 		}
 		seriesByTarget[s] = append(seriesByTarget[s], series)
 	}
@@ -767,7 +773,7 @@ func newRequestContext(ctx context.Context, req *models.Req, consolidator consol
 		// if the series has some excess at the end, it may aggregate into a bucket with a timestamp out of the desired range.
 		// for example: imagine we take the case from above, and the user specified a `to` of 115.
 		// a native 30s series would end with point 90. We should not include any points that would go into an aggregation bucket with timestamp higher than 90.
-		// (such as 100 or 110 which would technically be allowed by the `to` specification)
+		// (such as 100 or 110 which would technically be allowed by the `to` specification but land in the bucket with ts=120 which is out of bounds)
 		// so the proper to value is the highest value that does not result in points going into an out-of-bounds bucket.
 
 		// example: for 10s data (note that the 2 last colums should always match!)

api/dataprocessor_test.go

@@ -390,7 +390,7 @@ func TestGetSeriesFixed(t *testing.T) {
 				metric.Add(20+offset, 30) // this point will always be quantized to 30, so it should be selected
 				metric.Add(30+offset, 40) // this point will always be quantized to 40
 				metric.Add(40+offset, 50) // this point will always be quantized to 50
-				req := models.NewReq(id, "", "", from, to, 1000, 10, consolidation.Avg, 0, cluster.Manager.ThisNode(), 0, 0)
+				req := models.NewReq(id, "", "", from, to, 1000, 10, 0, consolidation.Avg, 0, cluster.Manager.ThisNode(), 0, 0)
 				req.Archive = 0
 				req.ArchInterval = 10
 				points, err := srv.getSeriesFixed(test.NewContext(), &models.StorageStats{}, req, consolidation.None)
@@ -582,7 +582,7 @@ func TestGetSeriesFixedVariableOutInterval(t *testing.T) {
 			metric.Add(dataPoint.Ts, dataPoint.Val)
 		}
 
-		req := models.NewReq(id, "", "", testCase.from, testCase.to, 1000, testCase.archInterval, consolidation.Avg, 0, cluster.Manager.ThisNode(), 0, 0)
+		req := models.NewReq(id, "", "", testCase.from, testCase.to, 1000, testCase.archInterval, 0, consolidation.Avg, 0, cluster.Manager.ThisNode(), 0, 0)
 		req.Archive = 0
 		req.ArchInterval = testCase.archInterval
 		req.OutInterval = testCase.outInterval
@@ -598,19 +598,10 @@ func TestGetSeriesFixedVariableOutInterval(t *testing.T) {
 }
 
 func reqRaw(key schema.MKey, from, to, maxPoints, rawInterval uint32, consolidator consolidation.Consolidator, schemaId, aggId uint16) models.Req {
-	req := models.NewReq(key, "", "", from, to, maxPoints, rawInterval, consolidator, 0, cluster.Manager.ThisNode(), schemaId, aggId)
+	req := models.NewReq(key, "", "", from, to, maxPoints, rawInterval, 0, consolidator, 0, cluster.Manager.ThisNode(), schemaId, aggId)
 	req.Archive = 0
 	return req
 }
-func reqOut(key schema.MKey, from, to, maxPoints, rawInterval uint32, consolidator consolidation.Consolidator, schemaId, aggId uint16, archive uint8, archInterval, ttl, outInterval, aggNum uint32) models.Req {
-	req := models.NewReq(key, "", "", from, to, maxPoints, rawInterval, consolidator, 0, cluster.Manager.ThisNode(), schemaId, aggId)
-	req.Archive = archive
-	req.ArchInterval = archInterval
-	req.TTL = ttl
-	req.OutInterval = outInterval
-	req.AggNum = aggNum
-	return req
-}
 
 func TestMergeSeries(t *testing.T) {
 	out := make([]models.Series, 0)

api/graphite.go

@@ -225,7 +225,13 @@ func (s *Server) renderMetrics(ctx *middleware.Context, request models.GraphiteR
 		// as graphite needs high-res data to perform its processing.
 		mdp = 0
 	}
-	plan, err := expr.NewPlan(exprs, fromUnix, toUnix, mdp, stable, nil)
+
+	opts, err := optimizations.ApplyUserPrefs(request.Optimizations)
+	if err != nil {
+		response.Write(ctx, response.NewError(http.StatusBadRequest, err.Error()))
+		return
+	}
+	plan, err := expr.NewPlan(exprs, fromUnix, toUnix, mdp, stable, opts)
 	if err != nil {
 		if fun, ok := err.(expr.ErrUnknownFunction); ok {
 			if request.NoProxy {
@@ -661,7 +667,7 @@ func (s *Server) executePlan(ctx context.Context, orgId uint32, plan expr.Plan)
 
 	minFrom := uint32(math.MaxUint32)
 	var maxTo uint32
-	var reqs []models.Req
+	reqs := NewReqMap()
 	metaTagEnrichmentData := make(map[string]tagquery.Tags)
 
 	// note that different patterns to query can have different from / to, so they require different index lookups
@@ -686,7 +692,7 @@ func (s *Server) executePlan(ctx context.Context, orgId uint32, plan expr.Plan)
 			if err != nil {
 				return nil, meta, err
 			}
-			series, err = s.clusterFindByTag(ctx, orgId, exprs, int64(r.From), maxSeriesPerReq-len(reqs), false)
+			series, err = s.clusterFindByTag(ctx, orgId, exprs, int64(r.From), maxSeriesPerReq-int(reqs.cnt), false)
 		} else {
 			series, err = s.findSeries(ctx, orgId, []string{r.Query}, int64(r.From))
 		}
@@ -718,9 +724,9 @@ func (s *Server) executePlan(ctx context.Context, orgId uint32, plan expr.Plan)
 						cons = closestAggMethod(consReq, mdata.Aggregations.Get(archive.AggId).AggregationMethod)
 					}
 
-					newReq := models.NewReq(
-						archive.Id, archive.NameWithTags(), r.Query, r.From, r.To, plan.MaxDataPoints, uint32(archive.Interval), cons, consReq, s.Node, archive.SchemaId, archive.AggId)
-					reqs = append(reqs, newReq)
+					newReq := r.ToModel()
+					newReq.Init(archive, cons, s.Node)
+					reqs.Add(newReq)
 				}
 
 				if tagquery.MetaTagSupport && len(metric.Defs) > 0 && len(metric.MetaTags) > 0 {
@@ -739,31 +745,35 @@ func (s *Server) executePlan(ctx context.Context, orgId uint32, plan expr.Plan)
 	default:
 	}
 
-	reqRenderSeriesCount.Value(len(reqs))
-	if len(reqs) == 0 {
+	reqRenderSeriesCount.ValueUint32(reqs.cnt)
+	if reqs.cnt == 0 {
 		return nil, meta, nil
 	}
 
-	meta.RenderStats.SeriesFetch = uint32(len(reqs))
+	meta.RenderStats.SeriesFetch = reqs.cnt
 
 	// note: if 1 series has a movingAvg that requires a long time range extension, it may push other reqs into another archive. can be optimized later
 	var err error
-	reqs, meta.RenderStats.PointsFetch, meta.RenderStats.PointsReturn, err = alignRequests(uint32(time.Now().Unix()), minFrom, maxTo, reqs)
+	var rp *ReqsPlan
+	rp, err = planRequests(uint32(time.Now().Unix()), minFrom, maxTo, reqs, plan.MaxDataPoints, maxPointsPerReqSoft, maxPointsPerReqHard)
 	if err != nil {
-		log.Errorf("HTTP Render alignReq error: %s", err.Error())
 		return nil, meta, err
 	}
+	meta.RenderStats.PointsFetch = rp.PointsFetch()
+	meta.RenderStats.PointsReturn = rp.PointsReturn(plan.MaxDataPoints)
+	reqsList := rp.List()
+
 	span := opentracing.SpanFromContext(ctx)
-	span.SetTag("num_reqs", len(reqs))
+	span.SetTag("num_reqs", len(reqsList))
 	span.SetTag("points_fetch", meta.RenderStats.PointsFetch)
 	span.SetTag("points_return", meta.RenderStats.PointsReturn)
 
-	for _, req := range reqs {
+	for _, req := range reqsList {
 		log.Debugf("HTTP Render %s - arch:%d archI:%d outI:%d aggN: %d from %s", req, req.Archive, req.ArchInterval, req.OutInterval, req.AggNum, req.Node.GetName())
 	}
 
 	a := time.Now()
-	out, err := s.getTargets(ctx, &meta.StorageStats, reqs)
+	out, err := s.getTargets(ctx, &meta.StorageStats, reqsList)
 	if err != nil {
 		log.Errorf("HTTP Render %s", err.Error())
 		return nil, meta, err
@@ -787,7 +797,7 @@ func (s *Server) executePlan(ctx context.Context, orgId uint32, plan expr.Plan)
 
 	data := make(map[expr.Req][]models.Series)
 	for _, serie := range out {
-		q := expr.NewReq(serie.QueryPatt, serie.QueryFrom, serie.QueryTo, serie.QueryCons)
+		q := expr.NewReqFromSerie(serie)
 		data[q] = append(data[q], serie)
 	}
 
@@ -1410,7 +1420,12 @@ func (s *Server) showPlan(ctx *middleware.Context, request models.GraphiteRender
 	stable := request.Process == "stable"
 	mdp := request.MaxDataPoints
 
-	plan, err := expr.NewPlan(exprs, fromUnix, toUnix, mdp, stable, nil)
+	opts, err := optimizations.ApplyUserPrefs(request.Optimizations)
+	if err != nil {
+		response.Write(ctx, response.NewError(http.StatusBadRequest, err.Error()))
+		return
+	}
+	plan, err := expr.NewPlan(exprs, fromUnix, toUnix, mdp, stable, opts)
 	if err != nil {
 		response.Write(ctx, response.NewError(http.StatusBadRequest, err.Error()))
 		return