fixup! rework request planning

Author: Dieterbe

api/graphite.go

@@ -749,14 +749,13 @@ func (s *Server) executePlan(ctx context.Context, orgId uint32, plan expr.Plan)
 	// 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
 	var rp *ReqsPlan
-	// TODO get rid of alignrequests and all "align" terminology
 	rp, err = planRequests(uint32(time.Now().Unix()), minFrom, maxTo, reqs, plan.MaxDataPoints)
 	meta.RenderStats.PointsFetch = rp.PointsFetch()
-	meta.RenderStats.PointsReturn = rp.PointsReturn()
+	meta.RenderStats.PointsReturn = rp.PointsReturn(plan.MaxDataPoints)
 	reqsList := rp.List()
 
 	if err != nil {
-		log.Errorf("HTTP Render alignReq error: %s", err.Error())
+		log.Errorf("HTTP Render planRequests error: %s", err.Error())
 		return nil, meta, err
 	}
 	span := opentracing.SpanFromContext(ctx)

api/graphite_req.go

@@ -75,20 +75,35 @@ func (rp ReqsPlan) PointsFetch() uint32 {
 	for _, r := range rp.single.mdpno {
 		cnt += r.PointsFetch()
 	}
-	// TODO also the groups
+	for _, split := range rp.pngroups {
+		for _, r := range split.mdpyes {
+			cnt += r.PointsFetch()
+		}
+		for _, r := range split.mdpno {
+			cnt += r.PointsFetch()
+		}
+	}
 	return cnt
 }
 
-// TODO implement this
-func (rp ReqsPlan) PointsReturn() uint32 {
+// PointsReturn estimates the amount of points that will be returned for this request
+// best effort: not aware of summarize(), aggregation functions, runtime normalization. but does account for runtime consolidation
+func (rp ReqsPlan) PointsReturn(planMDP uint32) uint32 {
 	var cnt uint32
 	for _, r := range rp.single.mdpyes {
-		cnt += r.PointsFetch()
+		cnt += r.PointsReturn(planMDP)
 	}
 	for _, r := range rp.single.mdpno {
-		cnt += r.PointsFetch()
+		cnt += r.PointsReturn(planMDP)
+	}
+	for _, split := range rp.pngroups {
+		for _, r := range split.mdpyes {
+			cnt += r.PointsReturn(planMDP)
+		}
+		for _, r := range split.mdpno {
+			cnt += r.PointsReturn(planMDP)
+		}
 	}
-	// TODO also the groups
 	return cnt
 }
 

api/models/request.go

@@ -50,7 +50,7 @@ type Req struct {
 type PNGroup uint64
 
 // Init initializes a request based on the metadata that we know of.
-// It sets all properties minus the ones that need request alignment
+// It sets all properties minus the ones that need request planning
 func (r *Req) Init(archive idx.Archive, cons consolidation.Consolidator, node cluster.Node) {
 	r.MKey = archive.Id
 	r.Target = archive.NameWithTags()
@@ -111,6 +111,20 @@ func (r *Req) AdjustTo(interval, from uint32, rets []conf.Retention) {
 func (r Req) PointsFetch() uint32 {
 	return (r.To - r.From) / r.ArchInterval
 }
+
+// PointsReturn estimates the amount of points that will be returned for this request
+// best effort: not aware of summarize(), runtime normalization. but does account for runtime consolidation
+func (r Req) PointsReturn(planMDP uint32) uint32 {
+	points := (r.To - r.From) / r.OutInterval
+	if planMDP > 0 && points > planMDP {
+		// note that we don't assign to req.AggNum here, because that's only for normalization.
+		// MDP runtime consolidation doesn't look at req.AggNum
+		aggNum := consolidation.AggEvery(points, planMDP)
+		points /= aggNum
+	}
+	return points
+}
+
 func (r Req) String() string {
 	return fmt.Sprintf("%s %d - %d (%s - %s) span:%ds. points <= %d. %s.", r.MKey.String(), r.From, r.To, util.TS(r.From), util.TS(r.To), r.To-r.From-1, r.MaxPoints, r.Consolidator)
 }
@@ -148,7 +162,7 @@ func (r Req) TraceLog(span opentracing.Span) {
 
 // Equals compares all fields of a to b for equality.
 // Except
-// * TTL (because alignRequests may change it)
+// * TTL (because planRequests may change it)
 //   for 100% correctness we may want to fix this in the future
 //   but for now, should be harmless since the field is not
 //   that important for archive fetching

api/query_engine.go

@@ -17,7 +17,8 @@ var (
 	reqRenderChosenArchive = stats.NewMeter32("api.request.render.chosen_archive", false)
 	// metric api.request.render.points_fetched is the number of points that need to be fetched for a /render request.
 	reqRenderPointsFetched = stats.NewMeter32("api.request.render.points_fetched", false)
-	// metric api.request.render.points_returned is the number of points the request will return.
+	// metric api.request.render.points_returned is the number of points the request will return
+	// best effort: not aware of summarize(), aggregation functions, runtime normalization. but does account for runtime consolidation
 	reqRenderPointsReturned = stats.NewMeter32("api.request.render.points_returned", false)
 
 	errUnSatisfiable   = response.NewError(404, "request cannot be satisfied due to lack of available retentions")
@@ -69,13 +70,13 @@ func planRequests(now, from, to uint32, reqs *ReqMap, planMDP uint32) (*ReqsPlan
 
 	for group, split := range rp.pngroups {
 		if len(split.mdpno) > 0 {
-			split.mdpno, ok = initialHighestResMulti(now, from, to, split.mdpno)
+			split.mdpno, ok = planHighestResMulti(now, from, to, split.mdpno)
 			if !ok {
 				return nil, errUnSatisfiable
 			}
 		}
 		if len(split.mdpyes) > 0 {
-			split.mdpyes, ok = initialLowestResForMDPMulti(now, from, to, split.mdpyes)
+			split.mdpyes, ok = planLowestResForMDPMulti(now, from, to, split.mdpyes)
 			if !ok {
 				return nil, errUnSatisfiable
 			}
@@ -84,30 +85,86 @@ func planRequests(now, from, to uint32, reqs *ReqMap, planMDP uint32) (*ReqsPlan
 	}
 	for i, req := range reqs.single {
 		if req.MaxPoints == 0 {
-			reqs.single[i], ok = initialHighestResSingle(now, from, to, req)
+			reqs.single[i], ok = planHighestResSingle(now, from, to, req)
 		} else {
-			reqs.single[i], ok = initialLowestResForMDPSingle(now, from, to, req)
+			reqs.single[i], ok = planLowestResForMDPSingle(now, from, to, req)
 		}
 		if !ok {
 			return nil, errUnSatisfiable
 		}
 	}
 
+	if maxPointsPerReqSoft > 0 {
+		// at this point, all MDP-optimizable series have already been optimized
+		// we can try to reduce the resolution of non-MDP-optimizable series
+		// if metrictank is already handling all, or most of your queries, then we have been able to determine
+		// MDP-optimizability very well. If the request came from Graphite, we have to assume it may run GR-functions.
+		// thus in the former case, we pretty much know that this is going to have an adverse effect on your queries,
+		// and you should probably not use this option, or we should even get rid of it.
+		// in the latter case though, it's quite likely we were too cautious and categorized many series as non-MDP
+		// optimizable whereas in reality they should be, so in that case this option is a welcome way to reduce the
+		// impact of big queries
+		// we could do two approaches: gradually reduce the interval of all series/groups being read, or just aggressively
+		// adjust one group at a time. The latter seems simpler, so for now we do just that.
+		if rp.PointsFetch() > uint32(maxPointsPerReqSoft) {
+			for group, split := range rp.pngroups {
+				if len(split.mdpno) > 0 {
+					split.mdpno, ok = planLowestResForMDPMulti(now, from, to, split.mdpno)
+					if !ok {
+						return nil, errUnSatisfiable
+					}
+					rp.pngroups[group] = split
+					if rp.PointsFetch() <= uint32(maxPointsPerReqSoft) {
+						goto HonoredSoft
+					}
+				}
+			}
+			for i, req := range reqs.single {
+				if req.MaxPoints == 0 {
+					reqs.single[i], ok = planLowestResForMDPSingle(now, from, to, req)
+				}
+				if !ok {
+					return nil, errUnSatisfiable
+				}
+				// for every 10 requests we adjusted, check if we honor soft now.
+				// note that there may be thousands of requests
+				if i%10 == 9 {
+					if rp.PointsFetch() <= uint32(maxPointsPerReqSoft) {
+						goto HonoredSoft
+					}
+				}
+			}
+		}
+	}
+HonoredSoft:
+
+	if int(rp.PointsFetch()) > maxPointsPerReqHard {
+		return nil, errMaxPointsPerReq
+
+	}
+
+	// send out some metrics and we're done!
+	for _, r := range rp.single.mdpyes {
+		reqRenderChosenArchive.ValueUint32(uint32(r.Archive))
+	}
+	for _, r := range rp.single.mdpno {
+		reqRenderChosenArchive.ValueUint32(uint32(r.Archive))
+	}
+	for _, split := range rp.pngroups {
+		for _, r := range split.mdpyes {
+			reqRenderChosenArchive.ValueUint32(uint32(r.Archive))
+		}
+		for _, r := range split.mdpno {
+			reqRenderChosenArchive.ValueUint32(uint32(r.Archive))
+		}
+	}
 	reqRenderPointsFetched.ValueUint32(rp.PointsFetch())
-	reqRenderPointsReturned.ValueUint32(rp.PointsReturn())
+	reqRenderPointsReturned.ValueUint32(rp.PointsReturn(planMDP))
 
 	return &rp, nil
 }
 
-/*
-		// requests in the same PNGroup can possibly receive a further tweak:
-		// if they currently require runtime normalization, we may instead be able
-		// to simply read from rollup archives and avoid runtime normalization
-		reqs, _, _, _ := alignRequests(now, from, to, groupReqs)
-	}
-*/
-
-func initialHighestResSingle(now, from, to uint32, req models.Req) (models.Req, bool) {
+func planHighestResSingle(now, from, to uint32, req models.Req) (models.Req, bool) {
 	rets := getRetentions(req)
 	minTTL := now - from
 	var ok bool
@@ -126,7 +183,7 @@ func initialHighestResSingle(now, from, to uint32, req models.Req) (models.Req,
 	return req, ok
 }
 
-func initialLowestResForMDPSingle(now, from, to uint32, req models.Req) (models.Req, bool) {
+func planLowestResForMDPSingle(now, from, to uint32, req models.Req) (models.Req, bool) {
 	rets := getRetentions(req)
 	var ok bool
 	for i := len(rets) - 1; i >= 0; i-- {
@@ -142,7 +199,7 @@ func initialLowestResForMDPSingle(now, from, to uint32, req models.Req) (models.
 	}
 	return req, ok
 }
-func initialHighestResMulti(now, from, to uint32, reqs []models.Req) ([]models.Req, bool) {
+func planHighestResMulti(now, from, to uint32, reqs []models.Req) ([]models.Req, bool) {
 	minTTL := now - from
 
 	var listIntervals []uint32
@@ -184,7 +241,7 @@ func initialHighestResMulti(now, from, to uint32, reqs []models.Req) ([]models.R
 }
 
 // note: we can assume all reqs have the same MDP.
-func initialLowestResForMDPMulti(now, from, to uint32, reqs []models.Req) ([]models.Req, bool) {
+func planLowestResForMDPMulti(now, from, to uint32, reqs []models.Req) ([]models.Req, bool) {
 	var ok bool
 	minTTL := now - from
 
@@ -250,29 +307,26 @@ func initialLowestResForMDPMulti(now, from, to uint32, reqs []models.Req) ([]mod
 			}
 		}
 	}
-	// now we finally found our optimal interval.
+	// now we finally found our optimal interval that we want to use.
 	// plan all our requests so that they result in the common output interval.
 	for i := range reqs {
 		req := &reqs[i]
-		req.AdjustTo(interval, from, getRetentions(*req))
-	}
-
-	return reqs, ok
-}
-
-/*
-func alignRequests(now, from, to uint32, reqs []models.Req) ([]models.Req, uint32, uint32, error) {
-	var pointsFetch uint32
-
-	pointsPerSerie := tsRange / interval
+		rets := getRetentions(*req)
+		for i := len(rets); i >= 0; i-- {
+			ret := rets[i]
+			if ret.Ready <= from && req.TTL >= minTTL {
+				if uint32(ret.SecondsPerPoint) == interval {
+					req.Plan(i, ret)
+					break
+				}
+				if interval%uint32(ret.SecondsPerPoint) == 0 {
+					req.Plan(i, ret)
+					req.PlanNormalization(interval)
+					break
+				}
+			}
+		}
 
-	// TODO series are not same resolution, need to account for separate intervals
-	if planMDP > 0 && pointsPerSerie > planMDP {
-		// note that we don't assign to req.AggNum here, because that's only for normalization.
-		// MDP runtime consolidation doesn't look at req.AggNum
-		aggNum := consolidation.AggEvery(pointsPerSerie, reqs[0].MaxPoints)
-		pointsPerSerie /= aggNum
+		return reqs, ok
 	}
-
 }
-*/

devdocs/expr.md

@@ -126,7 +126,7 @@ So:
    - consolidateBy setting defined closest to the leaf without a special* function in between the setting and the leaf, if available
    - determined via storage-aggregation.conf (defaults to average)
 3) at execution time, the consolidation settings encountered in consolidateBy calls travel up to the root because it is configured on the series, which is passed through the various layers of processing until it hits the root and the output step.  This becomes useful in two cases:
-   - when series need to be normalized at runtime, e.g. for sumSeries or divideSeries with series that have different steps; they need to be normalized (consolidated) so that the series get a compatible step, and the default of "avg" may not suffice.  (note that right now we have alignRequests which normalizes all series at fetch time, which can actually be a bit too eager, because some requests can use multiple targets with different processing - e.g. feed two different series into summarize(), so we actually don't need to normalize at runtime, but in the future we should make this better - TODO)
+   - when series need to be normalized at runtime, e.g. for sumSeries or divideSeries with series that have different steps; they need to be normalized (consolidated) so that the series get a compatible step, and the default of "avg" may not suffice.  (note that right now we have alignRequests which normalizes all series at fetch time, which can actually be a bit too eager, because some requests can use multiple targets with different processing - e.g. feed two different series into summarize(), so we actually don't need to normalize at runtime, but in the future we should make this better - TODO THIS IS OUT OF DATE) 
    - when returning data back to the user via a json response and whatnot, we can consolidate down using the method requested by the user (or average, if not specified). Likewise here, when the setting encounters a special* function while traveling up to the root, the consolidation value is reset to the default (average)
    Note: some functions combine multiple series into a new one (e.g. sumSeries, avgSeries, ...). Your input series may use different consolidateBy settings, some may be explicitly specified while others are not.  In this scenario, the output series will be given the first explicitly defined consolidateBy found by iterating the inputs, or the first default otherwise.
 

devdocs/maxdatapoints.txt

@@ -21,4 +21,4 @@ mdp set from GET param, but 0 if came from graphite
                     -> executePlan() models.NewReq() -> models.Req.MaxPoints
                         -> planRequests(): used for MDP-optimization
             -> plan.MaxDatapoints used for final runtime consolidation
-	    -> and also used in alignRequests() for reporting
+	    -> and also used in planRequests() for reporting

devdocs/render-request-handling.md

@@ -12,8 +12,8 @@
   * finds all series by fanning out the query patterns to all other shards. 
     this gives basically idx.Node back. has the path, leaf, metricdefinition, schema/aggregation(rollup) settings, for each series, as well as on which node it can be found.
   * construct models.Req objects for each serie. this uses the MKey to identify series, also sets from/to, maxdatapoints, etc.
-  * `alignRequests`: this looks at all models.Req objects and aligns them to a common step.
-    it selects the archive to use, consolidator settings etc (see NOTES in expr directory for more info)
+  * `planRequests`: this plans at all models.Req objects, which means decide which archive to read from, whether to apply normalization, etc
+    (see NOTES in expr directory for more info)
   * `getTargets`: gets the data from the local node and peer nodes based on the models.Req objects
   * `mergeSeries`: if there's multiple series with same name/tags, from, to and consolidator (e.g. because there's multiple series because users switched intervals), merge them together into one series
   * Sort each merged series so that the output of a function is well-defined and repeatable.

docs/consolidation.md

@@ -36,7 +36,7 @@ This further reduces data at runtime on an as-needed basis.
 It supports min, max, sum, average.
 
 
-## The request alignment algorithm
+## The request planning algorithm. OUT OF DATE AS OF https://github.com/grafana/metrictank/pull/951
 
 Metrictank uses a function called `alignRequests` which will:
 

docs/http-api.md

@@ -202,7 +202,7 @@ Each lineage section has these fields:
 | schema-retentions      | Retentions defined in storage-schemas.conf                                                                     |
 | archive-read           | Which archive was read as defined in the retentions. (0 means raw, 1 first rollup, etc)                        |
 | archive-interval       | The native interval of the archive that was read                                                               |
-| aggnum-norm            | If >1, number of points aggregated together per point, as part of normalization (series alignment)             |
+| aggnum-norm            | If >1, number of points aggregated together per point, as part of normalization                                |
 | aggnum-rc              | If >1, number of points aggregated together per output point, as part of runtime consolidation (MaxDataPoints) |
 | consolidator-normfetch | Consolidator used for normalization (if aggnum-norm > 1) and which rollup was read (if archive-read > 0)       |
 | consolidator-rc        | Consolidator used for runtime consolidation (MaxDataPoints) (if aggnum-rc > 1)                                 |

docs/render-path

@@ -88,6 +88,6 @@ such that they can be used together (for aggregating, merging, etc)
 
 
 TODO talk about
-alignRequests -> getTargets -> mergeSeries -> sort Series -> plan.Run (executes functions and does MDP consolidation with nudging)
+planRequests -> getTargets -> mergeSeries -> sort Series -> plan.Run (executes functions and does MDP consolidation with nudging)
 
 talk more about what happens at each step, how data is manipulated etc