in promql/engine.go [1021:1343]
func (ev *evaluator) eval(expr Expr) Value {
// This is the top-level evaluation method.
// Thus, we check for timeout/cancellation here.
if err := contextDone(ev.ctx, "expression evaluation"); err != nil {
ev.error(err)
}
numSteps := int((ev.endTimestamp-ev.startTimestamp)/ev.interval) + 1
switch e := expr.(type) {
case *AggregateExpr:
unwrapParenExpr(&e.Param)
if s, ok := e.Param.(*StringLiteral); ok {
return ev.rangeEval(func(v []Value, enh *EvalNodeHelper) Vector {
return ev.aggregation(e.Op, e.Grouping, e.Without, s.Val, v[0].(Vector), enh)
}, e.Expr)
}
return ev.rangeEval(func(v []Value, enh *EvalNodeHelper) Vector {
var param float64
if e.Param != nil {
param = v[0].(Vector)[0].V
}
return ev.aggregation(e.Op, e.Grouping, e.Without, param, v[1].(Vector), enh)
}, e.Param, e.Expr)
case *Call:
if e.Func.Name == "timestamp" {
// Matrix evaluation always returns the evaluation time,
// so this function needs special handling when given
// a vector selector.
vs, ok := e.Args[0].(*VectorSelector)
if ok {
return ev.rangeEval(func(v []Value, enh *EvalNodeHelper) Vector {
return e.Func.Call([]Value{ev.vectorSelector(vs, enh.ts)}, e.Args, enh)
})
}
}
// Check if the function has a matrix argument.
var matrixArgIndex int
var matrixArg bool
for i := range e.Args {
unwrapParenExpr(&e.Args[i])
a := e.Args[i]
if _, ok := a.(*MatrixSelector); ok {
matrixArgIndex = i
matrixArg = true
break
}
// SubqueryExpr can be used in place of MatrixSelector.
if subq, ok := a.(*SubqueryExpr); ok {
matrixArgIndex = i
matrixArg = true
// Replacing SubqueryExpr with MatrixSelector.
e.Args[i] = ev.evalSubquery(subq)
break
}
}
if !matrixArg {
// Does not have a matrix argument.
return ev.rangeEval(func(v []Value, enh *EvalNodeHelper) Vector {
return e.Func.Call(v, e.Args, enh)
}, e.Args...)
}
inArgs := make([]Value, len(e.Args))
// Evaluate any non-matrix arguments.
otherArgs := make([]Matrix, len(e.Args))
otherInArgs := make([]Vector, len(e.Args))
for i, e := range e.Args {
if i != matrixArgIndex {
otherArgs[i] = ev.eval(e).(Matrix)
otherInArgs[i] = Vector{Sample{}}
inArgs[i] = otherInArgs[i]
}
}
sel := e.Args[matrixArgIndex].(*MatrixSelector)
selVS := sel.VectorSelector.(*VectorSelector)
checkForSeriesSetExpansion(ev.ctx, sel)
mat := make(Matrix, 0, len(selVS.series)) // Output matrix.
offset := durationMilliseconds(selVS.Offset)
selRange := durationMilliseconds(sel.Range)
stepRange := selRange
if stepRange > ev.interval {
stepRange = ev.interval
}
// Reuse objects across steps to save memory allocations.
points := getPointSlice(16)
inMatrix := make(Matrix, 1)
inArgs[matrixArgIndex] = inMatrix
enh := &EvalNodeHelper{out: make(Vector, 0, 1)}
// Process all the calls for one time series at a time.
it := storage.NewBuffer(selRange)
for i, s := range selVS.series {
points = points[:0]
it.Reset(s.Iterator())
ss := Series{
// For all range vector functions, the only change to the
// output labels is dropping the metric name so just do
// it once here.
Metric: dropMetricName(selVS.series[i].Labels()),
Points: getPointSlice(numSteps),
}
inMatrix[0].Metric = selVS.series[i].Labels()
for ts, step := ev.startTimestamp, -1; ts <= ev.endTimestamp; ts += ev.interval {
step++
// Set the non-matrix arguments.
// They are scalar, so it is safe to use the step number
// when looking up the argument, as there will be no gaps.
for j := range e.Args {
if j != matrixArgIndex {
otherInArgs[j][0].V = otherArgs[j][0].Points[step].V
}
}
maxt := ts - offset
mint := maxt - selRange
// Evaluate the matrix selector for this series for this step.
points = ev.matrixIterSlice(it, mint, maxt, points)
if len(points) == 0 {
continue
}
inMatrix[0].Points = points
enh.ts = ts
// Make the function call.
outVec := e.Func.Call(inArgs, e.Args, enh)
enh.out = outVec[:0]
if len(outVec) > 0 {
ss.Points = append(ss.Points, Point{V: outVec[0].Point.V, T: ts})
}
// Only buffer stepRange milliseconds from the second step on.
it.ReduceDelta(stepRange)
}
if len(ss.Points) > 0 {
if ev.currentSamples < ev.maxSamples {
mat = append(mat, ss)
ev.currentSamples += len(ss.Points)
} else {
ev.error(ErrTooManySamples(env))
}
} else {
putPointSlice(ss.Points)
}
}
putPointSlice(points)
// The absent_over_time function returns 0 or 1 series. So far, the matrix
// contains multiple series. The following code will create a new series
// with values of 1 for the timestamps where no series has value.
if e.Func.Name == "absent_over_time" {
steps := int(1 + (ev.endTimestamp-ev.startTimestamp)/ev.interval)
// Iterate once to look for a complete series.
for _, s := range mat {
if len(s.Points) == steps {
return Matrix{}
}
}
found := map[int64]struct{}{}
for i, s := range mat {
for _, p := range s.Points {
found[p.T] = struct{}{}
}
if i > 0 && len(found) == steps {
return Matrix{}
}
}
newp := make([]Point, 0, steps-len(found))
for ts := ev.startTimestamp; ts <= ev.endTimestamp; ts += ev.interval {
if _, ok := found[ts]; !ok {
newp = append(newp, Point{T: ts, V: 1})
}
}
return Matrix{
Series{
Metric: createLabelsForAbsentFunction(e.Args[0]),
Points: newp,
},
}
}
if mat.ContainsSameLabelset() {
ev.errorf("vector cannot contain metrics with the same labelset")
}
return mat
case *ParenExpr:
return ev.eval(e.Expr)
case *UnaryExpr:
mat := ev.eval(e.Expr).(Matrix)
if e.Op == SUB {
for i := range mat {
mat[i].Metric = dropMetricName(mat[i].Metric)
for j := range mat[i].Points {
mat[i].Points[j].V = -mat[i].Points[j].V
}
}
if mat.ContainsSameLabelset() {
ev.errorf("vector cannot contain metrics with the same labelset")
}
}
return mat
case *BinaryExpr:
switch lt, rt := e.LHS.Type(), e.RHS.Type(); {
case lt == ValueTypeScalar && rt == ValueTypeScalar:
return ev.rangeEval(func(v []Value, enh *EvalNodeHelper) Vector {
val := scalarBinop(e.Op, v[0].(Vector)[0].Point.V, v[1].(Vector)[0].Point.V)
return append(enh.out, Sample{Point: Point{V: val}})
}, e.LHS, e.RHS)
case lt == ValueTypeVector && rt == ValueTypeVector:
switch e.Op {
case LAND:
return ev.rangeEval(func(v []Value, enh *EvalNodeHelper) Vector {
return ev.VectorAnd(v[0].(Vector), v[1].(Vector), e.VectorMatching, enh)
}, e.LHS, e.RHS)
case LOR:
return ev.rangeEval(func(v []Value, enh *EvalNodeHelper) Vector {
return ev.VectorOr(v[0].(Vector), v[1].(Vector), e.VectorMatching, enh)
}, e.LHS, e.RHS)
case LUNLESS:
return ev.rangeEval(func(v []Value, enh *EvalNodeHelper) Vector {
return ev.VectorUnless(v[0].(Vector), v[1].(Vector), e.VectorMatching, enh)
}, e.LHS, e.RHS)
default:
return ev.rangeEval(func(v []Value, enh *EvalNodeHelper) Vector {
return ev.VectorBinop(e.Op, v[0].(Vector), v[1].(Vector), e.VectorMatching, e.ReturnBool, enh)
}, e.LHS, e.RHS)
}
case lt == ValueTypeVector && rt == ValueTypeScalar:
return ev.rangeEval(func(v []Value, enh *EvalNodeHelper) Vector {
return ev.VectorscalarBinop(e.Op, v[0].(Vector), Scalar{V: v[1].(Vector)[0].Point.V}, false, e.ReturnBool, enh)
}, e.LHS, e.RHS)
case lt == ValueTypeScalar && rt == ValueTypeVector:
return ev.rangeEval(func(v []Value, enh *EvalNodeHelper) Vector {
return ev.VectorscalarBinop(e.Op, v[1].(Vector), Scalar{V: v[0].(Vector)[0].Point.V}, true, e.ReturnBool, enh)
}, e.LHS, e.RHS)
}
case *NumberLiteral:
return ev.rangeEval(func(v []Value, enh *EvalNodeHelper) Vector {
return append(enh.out, Sample{Point: Point{V: e.Val}})
})
case *VectorSelector:
checkForSeriesSetExpansion(ev.ctx, e)
mat := make(Matrix, 0, len(e.series))
it := storage.NewBuffer(durationMilliseconds(LookbackDelta))
for i, s := range e.series {
it.Reset(s.Iterator())
ss := Series{
Metric: e.series[i].Labels(),
Points: getPointSlice(numSteps),
}
for ts := ev.startTimestamp; ts <= ev.endTimestamp; ts += ev.interval {
_, v, ok := ev.vectorSelectorSingle(it, e, ts)
if ok {
if ev.currentSamples < ev.maxSamples {
ss.Points = append(ss.Points, Point{V: v, T: ts})
ev.currentSamples++
} else {
ev.error(ErrTooManySamples(env))
}
}
}
if len(ss.Points) > 0 {
mat = append(mat, ss)
} else {
putPointSlice(ss.Points)
}
}
return mat
case *MatrixSelector:
if ev.startTimestamp != ev.endTimestamp {
panic(errors.New("cannot do range evaluation of matrix selector"))
}
return ev.matrixSelector(e)
case *SubqueryExpr:
offsetMillis := durationToInt64Millis(e.Offset)
rangeMillis := durationToInt64Millis(e.Range)
newEv := &evaluator{
endTimestamp: ev.endTimestamp - offsetMillis,
interval: ev.defaultEvalInterval,
ctx: ev.ctx,
currentSamples: ev.currentSamples,
maxSamples: ev.maxSamples,
defaultEvalInterval: ev.defaultEvalInterval,
logger: ev.logger,
}
if e.Step != 0 {
newEv.interval = durationToInt64Millis(e.Step)
}
// Start with the first timestamp after (ev.startTimestamp - offset - range)
// that is aligned with the step (multiple of 'newEv.interval').
newEv.startTimestamp = newEv.interval * ((ev.startTimestamp - offsetMillis - rangeMillis) / newEv.interval)
if newEv.startTimestamp < (ev.startTimestamp - offsetMillis - rangeMillis) {
newEv.startTimestamp += newEv.interval
}
res := newEv.eval(e.Expr)
ev.currentSamples = newEv.currentSamples
return res
case *StringLiteral:
return String{V: e.Val, T: ev.startTimestamp}
}
panic(errors.Errorf("unhandled expression of type: %T", expr))
}