graphite expression parser

mostly just parsing of expressions

Author: Dieterbe

.gitignore

@@ -3,6 +3,7 @@
 build/
 /metrictank
 /cmd/mt-aggs-explain/mt-aggs-explain
+/cmd/mt-explain/mt-explain
 /cmd/mt-index-cat/mt-index-cat
 /cmd/mt-index-migrate/mt-index-migrate
 /cmd/mt-index-migrate-050-to-054/mt-index-migrate-050-to-054

api/graphite.go

@@ -2,6 +2,7 @@ package api
 
 import (
 	"errors"
+	"math"
 	"net/http"
 	"sort"
 	"strings"
@@ -14,9 +15,11 @@ import (
 	"github.com/raintank/metrictank/api/response"
 	"github.com/raintank/metrictank/cluster"
 	"github.com/raintank/metrictank/consolidation"
+	"github.com/raintank/metrictank/expr"
 	"github.com/raintank/metrictank/idx"
 	"github.com/raintank/metrictank/mdata"
 	"github.com/raintank/metrictank/stats"
+	"github.com/raintank/metrictank/util"
 	"github.com/raintank/worldping-api/pkg/log"
 )
 
@@ -40,34 +43,6 @@ type Series struct {
 	Node    cluster.Node
 }
 
-func parseTarget(target string) (string, string, error) {
-	var consolidateBy string
-	// yes, i am aware of the arguably grossness of the below.
-	// however, it is solid based on the documented allowed input format.
-	// once we need to support several functions, we can implement
-	// a proper expression parser
-	if strings.HasPrefix(target, "consolidateBy(") {
-		var q1, q2 int
-		t := target
-		if t[len(t)-2:] == "')" && (strings.Contains(t, ",'") || strings.Contains(t, ", '")) && strings.Count(t, "'") == 2 {
-			q1 = strings.Index(t, "'")
-			q2 = strings.LastIndex(t, "'")
-		} else if t[len(t)-2:] == "\")" && (strings.Contains(t, ",\"") || strings.Contains(t, ", \"")) && strings.Count(t, "\"") == 2 {
-			q1 = strings.Index(t, "\"")
-			q2 = strings.LastIndex(t, "\"")
-		} else {
-			return "", "", response.NewError(http.StatusBadRequest, "target parse error")
-		}
-		consolidateBy = t[q1+1 : q2]
-		err := consolidation.Validate(consolidateBy)
-		if err != nil {
-			return "", "", err
-		}
-		target = t[strings.Index(t, "(")+1 : strings.LastIndex(t, ",")]
-	}
-	return target, consolidateBy, nil
-}
-
 func (s *Server) findSeries(orgId int, patterns []string, seenAfter int64) ([]Series, error) {
 	peers := cluster.MembersForQuery()
 	log.Debug("HTTP findSeries for %v across %d instances", patterns, len(peers))
@@ -189,112 +164,37 @@ func (s *Server) renderMetrics(ctx *middleware.Context, request models.GraphiteR
 		return
 	}
 
-	reqs := make([]models.Req, 0)
-
-	patterns := make([]string, 0)
-	type locatedDef struct {
-		def  idx.Archive
-		node cluster.Node
-	}
-
-	//locatedDefs[<pattern>][<def.id>]locatedDef
-	locatedDefs := make(map[string]map[string]locatedDef)
-	//targetForPattern[<pattern>]<target>
-	targetForPattern := make(map[string]string)
-	for _, target := range targets {
-		pattern, _, err := parseTarget(target)
-		if err != nil {
-			ctx.Error(http.StatusBadRequest, err.Error())
-			return
-		}
-		patterns = append(patterns, pattern)
-		targetForPattern[pattern] = target
-		locatedDefs[pattern] = make(map[string]locatedDef)
-
-	}
-
-	series, err := s.findSeries(ctx.OrgId, patterns, int64(fromUnix))
+	exprs, err := expr.ParseMany(targets)
 	if err != nil {
-		response.Write(ctx, response.WrapError(err))
-	}
-
-	for _, s := range series {
-		for _, metric := range s.Series {
-			if !metric.Leaf {
-				continue
-			}
-			for _, def := range metric.Defs {
-				locatedDefs[s.Pattern][def.Id] = locatedDef{def, s.Node}
-			}
-		}
-	}
-
-	for pattern, ldefs := range locatedDefs {
-		for _, locdef := range ldefs {
-			archive := locdef.def
-			// set consolidator that will be used to normalize raw data before feeding into processing functions
-			// not to be confused with runtime consolidation which happens in the graphite api, after all processing.
-			fn := mdata.Aggregations.Get(archive.AggId).AggregationMethod[0]
-			consolidator := consolidation.Consolidator(fn) // we use the same number assignments so we can cast them
-			// target is like foo.bar or foo.* or consolidateBy(foo.*,'sum')
-			// pattern is like foo.bar or foo.*
-			// def.Name is like foo.concretebar
-			// so we want target to contain the concrete graphite name, potentially wrapped with consolidateBy().
-			target := strings.Replace(targetForPattern[pattern], pattern, archive.Name, -1)
-			reqs = append(reqs, models.NewReq(archive.Id, target, fromUnix, toUnix, request.MaxDataPoints, uint32(archive.Interval), consolidator, locdef.node, archive.SchemaId, archive.AggId))
-		}
-	}
-
-	reqRenderSeriesCount.Value(len(reqs))
-	reqRenderTargetCount.Value(len(request.Targets))
-
-	if len(reqs) == 0 {
-		if request.Format == "msgp" {
-			var series models.SeriesByTarget
-			response.Write(ctx, response.NewMsgp(200, series))
-		} else {
-			response.Write(ctx, response.NewJson(200, []string{}, ""))
-		}
+		ctx.Error(http.StatusBadRequest, err.Error())
 		return
 	}
 
-	if (toUnix - fromUnix) >= logMinDur {
-		log.Info("HTTP Render: INCOMING REQ %q from: %q, to: %q target cnt: %d, maxDataPoints: %d",
-			ctx.Req.Method, from, to, len(request.Targets), request.MaxDataPoints)
-	}
+	reqRenderTargetCount.Value(len(targets))
 
-	reqs, err = alignRequests(uint32(time.Now().Unix()), reqs)
+	plan, err := expr.NewPlan(exprs, fromUnix, toUnix, request.MaxDataPoints, nil)
 	if err != nil {
-		log.Error(3, "HTTP Render alignReq error: %s", err)
-		response.Write(ctx, response.WrapError(err))
+		ctx.Error(http.StatusBadRequest, err.Error())
 		return
 	}
 
-	if LogLevel < 2 {
-		for _, req := range reqs {
-			log.Debug("HTTP Render %s - arch:%d archI:%d outI:%d aggN: %d from %s", req, req.Archive, req.ArchInterval, req.OutInterval, req.AggNum, req.Node.Name)
-		}
-	}
-
-	out, err := s.getTargets(reqs)
+	out, err := s.executePlan(ctx.OrgId, plan)
 	if err != nil {
-		log.Error(3, "HTTP Render %s", err.Error())
-		response.Write(ctx, response.WrapError(err))
+		ctx.Error(http.StatusBadRequest, err.Error())
 		return
 	}
+	sort.Sort(models.SeriesByTarget(out))
 
-	merged := mergeSeries(out)
-	sort.Sort(models.SeriesByTarget(merged))
 	defer func() {
 		for _, serie := range out {
 			pointSlicePool.Put(serie.Datapoints[:0])
 		}
 	}()
 
 	if request.Format == "msgp" {
-		response.Write(ctx, response.NewMsgp(200, models.SeriesByTarget(merged)))
+		response.Write(ctx, response.NewMsgp(200, models.SeriesByTarget(out)))
 	} else {
-		response.Write(ctx, response.NewFastJson(200, models.SeriesByTarget(merged)))
+		response.Write(ctx, response.NewFastJson(200, models.SeriesByTarget(out)))
 	}
 }
 
@@ -557,3 +457,81 @@ func (s *Server) metricsDeleteRemote(orgId int, query string, peer cluster.Node)
 
 	return resp.DeletedDefs, nil
 }
+
+// note if you do something like sum(foo.*) and all of those metrics happen to be on another node,
+// we will collect all the indidividual series from the peer, and then sum here. that could be optimized
+func (s *Server) executePlan(orgId int, plan expr.Plan) ([]models.Series, error) {
+
+	type locatedDef struct {
+		def  idx.Archive
+		node cluster.Node
+	}
+	minFrom := uint32(math.MaxUint32)
+	var maxTo uint32
+	//locatedDefs[request][<def.id>]locatedDef
+	locatedDefs := make(map[expr.Req]map[string]locatedDef)
+
+	// note that different patterns to query can have different from / to, so they require different index lookups
+	// e.g. target=movingAvg(foo.*, "1h")&target=foo.*
+	// note that in this case we fetch foo.* twice. can be optimized later
+	for _, r := range plan.Reqs {
+		series, err := s.findSeries(orgId, []string{r.Query}, int64(r.From))
+		if err != nil {
+			return nil, err
+		}
+
+		locatedDefs[r] = make(map[string]locatedDef)
+
+		minFrom = util.Min(minFrom, r.From)
+		maxTo = util.Max(maxTo, r.To)
+
+		for _, s := range series {
+			for _, metric := range s.Series {
+				if !metric.Leaf {
+					continue
+				}
+				for _, def := range metric.Defs {
+					locatedDefs[r][def.Id] = locatedDef{def, s.Node}
+				}
+			}
+		}
+	}
+
+	var reqs []models.Req
+	for r, ldefs := range locatedDefs {
+		for _, locdef := range ldefs {
+			archive := locdef.def
+			// set consolidator that will be used to normalize raw data before feeding into processing functions
+			// not to be confused with runtime consolidation which happens after all processing.
+			fn := mdata.Aggregations.Get(archive.AggId).AggregationMethod[0]
+			consolidator := consolidation.Consolidator(fn) // we use the same number assignments so we can cast them
+			reqs = append(reqs, models.NewReq(
+				archive.Id, archive.Name, r.From, r.To, plan.MaxDataPoints, uint32(archive.Interval), consolidator, locdef.node, archive.SchemaId, archive.AggId))
+		}
+	}
+	reqRenderSeriesCount.Value(len(reqs))
+	if len(reqs) == 0 {
+		return nil, nil
+	}
+	// 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
+	reqs, err := alignRequests(uint32(time.Now().Unix()), minFrom, maxTo, reqs)
+	if err != nil {
+		log.Error(3, "HTTP Render alignReq error: %s", err)
+		return nil, err
+	}
+
+	if LogLevel < 2 {
+		for _, req := range reqs {
+			log.Debug("HTTP Render %s - arch:%d archI:%d outI:%d aggN: %d from %s", req, req.Archive, req.ArchInterval, req.OutInterval, req.AggNum, req.Node.Name)
+		}
+	}
+
+	out, err := s.getTargets(reqs)
+	if err != nil {
+		log.Error(3, "HTTP Render %s", err.Error())
+		return nil, err
+	}
+
+	merged := mergeSeries(out)
+	return merged, nil
+}

api/query_engine.go

@@ -26,8 +26,8 @@ var (
 // alignRequests updates the requests with all details for fetching, making sure all metrics are in the same, optimal interval
 // note: it is assumed that all requests have the same from & to.
 // also takes a "now" value which we compare the TTL against
-func alignRequests(now uint32, reqs []models.Req) ([]models.Req, error) {
-	tsRange := (reqs[0].To - reqs[0].From)
+func alignRequests(now, from, to uint32, reqs []models.Req) ([]models.Req, error) {
+	tsRange := to - from
 
 	var listIntervals []uint32
 	var seenIntervals = make(map[uint32]struct{})

api/query_engine_test.go

@@ -28,7 +28,7 @@ func testAlign(reqs []models.Req, retentions [][]conf.Retention, outReqs []model
 	}
 
 	mdata.Schemas = conf.NewSchemas(schemas)
-	out, err := alignRequests(now, reqs)
+	out, err := alignRequests(now, reqs[0].From, reqs[0].To, reqs)
 	if err != outErr {
 		t.Errorf("different err value expected: %v, got: %v", outErr, err)
 	}
@@ -430,7 +430,7 @@ func testMaxPointsPerReq(maxPointsSoft, maxPointsHard int, reqs []models.Req, t
 		}),
 	}})
 
-	out, err := alignRequests(30*day, reqs)
+	out, err := alignRequests(30*day, reqs[0].From, reqs[0].To, reqs)
 	maxPointsPerReqSoft = origMaxPointsPerReqSoft
 	maxPointsPerReqHard = origMaxPointsPerReqHard
 	return out, err
@@ -523,7 +523,7 @@ func BenchmarkAlignRequests(b *testing.B) {
 	})
 
 	for n := 0; n < b.N; n++ {
-		res, _ = alignRequests(14*24*3600, reqs)
+		res, _ = alignRequests(14*24*3600, 0, 3600*24*7, reqs)
 	}
 	result = res
 }

cmd/mt-explain/main.go

@@ -0,0 +1,25 @@
+package main
+
+import (
+	"fmt"
+
+	"github.com/davecgh/go-spew/spew"
+	"github.com/raintank/metrictank/expr"
+)
+
+func main() {
+	target := "movingAverage(sumSeries(foo.bar), '2min')"
+	exps, err := expr.ParseMany([]string{target})
+	if err != nil {
+		fmt.Println("Error while parsing:", err)
+		return
+	}
+	spew.Dump(exps)
+
+	plan, err := expr.NewPlan(exps, 1000, 1200, nil)
+	if err != nil {
+		fmt.Println("Error while planning", err)
+		return
+	}
+	spew.Dump(plan)
+}

expr/expr.go

@@ -0,0 +1,20 @@
+package expr
+
+type exprType int
+
+const (
+	etName exprType = iota // e.g. a metric query pattern like foo.bar or foo.*.baz
+	etFunc
+	etConst
+	etString
+)
+
+type expr struct {
+	target    string // the name of a etName, or func name for etFunc. not used for etConst
+	etype     exprType
+	val       float64 // for etConst
+	valStr    string  // for etString, and to store original input for etConst
+	args      []*expr // positional
+	namedArgs map[string]*expr
+	argString string // for etFunc: literal string of how all the args were specified
+}

expr/func_consolidateby.go

@@ -0,0 +1,22 @@
+package expr
+
+import "github.com/raintank/metrictank/consolidation"
+
+type FuncConsolidateBy struct {
+}
+
+func NewConsolidateBy() Func {
+	return FuncConsolidateBy{}
+}
+
+func (s FuncConsolidateBy) Signature() ([]argType, []argType) {
+	return []argType{seriesList, str}, []argType{seriesList}
+}
+
+func (s FuncConsolidateBy) Init(args []*expr) error {
+	return consolidation.Validate(args[1].valStr)
+}
+
+func (s FuncConsolidateBy) Depends(from, to uint32) (uint32, uint32) {
+	return from, to
+}