Skip to content

Commit

Permalink
Development (#70)
Browse files Browse the repository at this point in the history 
* IG 15394 - use new tsdb metric cache (#67)

* new tsdb addFast

* new AddFast

Co-authored-by: Dina Nimrodi <[email protected]>

* new tsdb 0.11.2 (#68)

Co-authored-by: Dina Nimrodi <[email protected]>

* fix step parsing in subQueries + fix inferring whether v3io needs to run aggregations (#66)

* validate development unstable

* Fix go.mod. go mod vendor. (#72)

Co-authored-by: Gal Topper <[email protected]>

Co-authored-by: Dina Nimrodi <[email protected]>
Co-authored-by: Tal Neiman <[email protected]>
Co-authored-by: ork <[email protected]>
Co-authored-by: Gal Topper <[email protected]>
Co-authored-by: Gal Topper <[email protected]>
  • Loading branch information
@dinal @talIguaz @orkiguazio
6 people authored Oct 12, 2020
1 parent 8a3b020 commit 7a632e7
Show file tree
Hide file tree
Showing 29 changed files with 982 additions and 330 deletions.
1 change: 1 addition & 0 deletions README.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -121,3 +121,4 @@ Apache License 2.0, see [LICENSE](https://github.com/prometheus/prometheus/blob/
[hub]: https://hub.docker.com/r/prom/prometheus/
[circleci]: https://circleci.com/gh/prometheus/prometheus
[quay]: https://quay.io/repository/prometheus/prometheus

6 changes: 3 additions & 3 deletions go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ require (
github.com/shurcooL/vfsgen v0.0.0-20181202132449-6a9ea43bcacd
github.com/soheilhy/cmux v0.1.4
github.com/stretchr/testify v1.4.0
github.com/v3io/v3io-tsdb v0.10.12
github.com/v3io/v3io-tsdb v0.11.2
golang.org/x/net v0.0.0-20190724013045-ca1201d0de80
golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45
golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a
Expand All @@ -61,8 +61,8 @@ require (

replace (
github.com/golang/lint => golang.org/x/lint v0.0.0-20190409202823-959b441ac422
github.com/v3io/frames => github.com/v3io/frames v0.7.36
github.com/v3io/v3io-tsdb => github.com/v3io/v3io-tsdb v0.10.12
github.com/v3io/frames => github.com/v3io/frames v0.8.1
github.com/v3io/v3io-tsdb => github.com/v3io/v3io-tsdb v0.11.2
google.golang.org/grpc => google.golang.org/grpc v1.19.1
k8s.io/klog => github.com/simonpasquier/klog-gokit v0.1.0
)
8 changes: 4 additions & 4 deletions go.sum
View file
Original file line number Diff line number Diff line change
Expand Up @@ -388,12 +388,12 @@ github.com/tidwall/pretty v1.0.0 h1:HsD+QiTn7sK6flMKIvNmpqz1qrpP3Ps6jOKIKMooyg4=
github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
github.com/tinylib/msgp v1.1.1 h1:TnCZ3FIuKeaIy+F45+Cnp+caqdXGy4z74HvwXN+570Y=
github.com/tinylib/msgp v1.1.1/go.mod h1:+d+yLhGm8mzTaHzB+wgMYrodPfmZrzkirds8fDWklFE=
github.com/v3io/frames v0.7.36 h1:vWLJWbjvKMgWbCNrMj79wfKStcmvuAkh2UyfNGPIRfs=
github.com/v3io/frames v0.7.36/go.mod h1:qUW5rLv+DK4pC0t52685E9HxqDBg3VFgourbbYfRmaI=
github.com/v3io/frames v0.8.1 h1:wUWCX6RUEQ05SG3P9VtMBYh7gWrYNOiA81HcWXRXGZw=
github.com/v3io/frames v0.8.1/go.mod h1:qQnXBGO3uLKwN9HVcQ2ggHlIs8xT+200uJaWUgDocpk=
github.com/v3io/v3io-go v0.1.9 h1:etkrrRmrI++i8sxGfK/+13f41TxIMohYeZHwVUM62vw=
github.com/v3io/v3io-go v0.1.9/go.mod h1:5poBlcjZG5TiexRTYI44PE6tHzZz5Z60w+iS899pWtc=
github.com/v3io/v3io-tsdb v0.10.12 h1:+AfPHFJVxzpKtng0Hiz256ef9abzIwKwPuzB+0jG7w8=
github.com/v3io/v3io-tsdb v0.10.12/go.mod h1:kp586KxTfROIGwb/nzNxwDbX2Wterxro+HbiZHmK548=
github.com/v3io/v3io-tsdb v0.11.2 h1:knT+IwB+Bkd1aRPTIKltBnYrue1z+2EXmESJLkCexe8=
github.com/v3io/v3io-tsdb v0.11.2/go.mod h1:l+WxedjLmjy/TThj2vhSW/OnpRw8C7dOntIXHmM/a7I=
github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
github.com/valyala/fasthttp v1.2.0 h1:dzZJf2IuMiclVjdw0kkT+f9u4YdrapbNyGAN47E/qnk=
Expand Down
98 changes: 61 additions & 37 deletions promql/engine.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -464,14 +464,14 @@ func (ng *Engine) execEvalStmt(ctx context.Context, query *query, s *EvalStmt) (
if s.Start == s.End && s.Interval == 0 {
start := timeMilliseconds(s.Start)
evaluator := &evaluator{
startTimestamp: start,
endTimestamp: start,
interval: 1,
ctx: ctxInnerEval,
maxSamples: ng.maxSamplesPerQuery,
defaultEvalInterval: GetDefaultEvaluationInterval(),
logger: ng.logger,
useV3ioAggregations: querier.(*tsdb.V3ioPromQuerier).UseV3ioAggregations(),
startTimestamp: start,
endTimestamp: start,
interval: 1,
ctx: ctxInnerEval,
maxSamples: ng.maxSamplesPerQuery,
defaultEvalInterval: GetDefaultEvaluationInterval(),
logger: ng.logger,
isAlreadyV3IOAggregated: querier.(*tsdb.V3ioPromQuerier).IsAlreadyAggregated,
}
val, err := evaluator.Eval(s.Expr)
if err != nil {
Expand Down Expand Up @@ -507,14 +507,14 @@ func (ng *Engine) execEvalStmt(ctx context.Context, query *query, s *EvalStmt) (

// Range evaluation.
evaluator := &evaluator{
startTimestamp: timeMilliseconds(s.Start),
endTimestamp: timeMilliseconds(s.End),
interval: durationMilliseconds(s.Interval),
ctx: ctxInnerEval,
maxSamples: ng.maxSamplesPerQuery,
defaultEvalInterval: GetDefaultEvaluationInterval(),
logger: ng.logger,
useV3ioAggregations: querier.(*tsdb.V3ioPromQuerier).UseV3ioAggregations(),
startTimestamp: timeMilliseconds(s.Start),
endTimestamp: timeMilliseconds(s.End),
interval: durationMilliseconds(s.Interval),
ctx: ctxInnerEval,
maxSamples: ng.maxSamplesPerQuery,
defaultEvalInterval: GetDefaultEvaluationInterval(),
logger: ng.logger,
isAlreadyV3IOAggregated: querier.(*tsdb.V3ioPromQuerier).IsAlreadyAggregated,
}
val, err := evaluator.Eval(s.Expr)
if err != nil {
Expand Down Expand Up @@ -556,6 +556,7 @@ func (ng *Engine) cumulativeSubqueryOffset(path []Node) time.Duration {
func (ng *Engine) populateSeries(ctx context.Context, q storage.Queryable, s *EvalStmt) (storage.Querier, storage.Warnings, error) {
var maxOffset time.Duration
var aggregationWindow int64
var subQueryStep int64

Inspect(s.Expr, func(node Node, path []Node) error {
subqOffset := ng.cumulativeSubqueryOffset(path)
Expand All @@ -568,13 +569,17 @@ func (ng *Engine) populateSeries(ctx context.Context, q storage.Queryable, s *Ev
maxOffset = n.Offset + LookbackDelta + subqOffset
}
case *MatrixSelector:
aggregationWindow = n.Range.Nanoseconds() / 1000000
aggregationWindow = durationToInt64Millis(n.Range)
if maxOffset < n.Range+subqOffset {
maxOffset = n.Range + subqOffset
}
if n.Offset+n.Range+subqOffset > maxOffset {
maxOffset = n.Offset + n.Range + subqOffset
}
case *SubqueryExpr:
// Save the step if it is provided in a subquery rather than as an interval
// Example query: `sum(metric)[1h:10m]` -> step=10m
subQueryStep = durationToInt64Millis(n.Step)
}
return nil
})
Expand All @@ -591,10 +596,16 @@ func (ng *Engine) populateSeries(ctx context.Context, q storage.Queryable, s *Ev
Inspect(s.Expr, func(node Node, path []Node) error {
var set storage.SeriesSet
var wrn storage.Warnings

// Get the Step from the sub query, in case the interval is not set.
step := durationToInt64Millis(s.Interval)
if step == 0 {
step = subQueryStep
}
params := &storage.SelectParams{
Start: timestamp.FromTime(s.Start),
End: timestamp.FromTime(s.End),
Step: durationToInt64Millis(s.Interval),
Step: step,
}

// We need to make sure we select the timerange selected by the subquery.
Expand All @@ -607,6 +618,9 @@ func (ng *Engine) populateSeries(ctx context.Context, q storage.Queryable, s *Ev

switch n := node.(type) {
case *VectorSelector:
// Validate if the current query can be aggregated via v3io
querier.(*tsdb.V3ioPromQuerier).UseAggregates = isV3ioEligibleQueryExpr(path)

params.Start = params.Start - durationMilliseconds(LookbackDelta)
params.Func = extractFuncFromPath(path)
params.By, params.Grouping = extractGroupsFromPath(path)
Expand All @@ -629,6 +643,9 @@ func (ng *Engine) populateSeries(ctx context.Context, q storage.Queryable, s *Ev
n.unexpandedSeriesSet = set

case *MatrixSelector:
// Validate if the current query can be aggregated via v3io
querier.(*tsdb.V3ioPromQuerier).UseAggregates = isV3ioEligibleQueryExpr(path)

params.Func = extractFuncFromPath(path)
params.Range = durationMilliseconds(n.Range)
// For all matrix queries we want to ensure that we have (end-start) + range selected
Expand Down Expand Up @@ -738,7 +755,7 @@ type evaluator struct {
defaultEvalInterval int64
logger log.Logger

useV3ioAggregations bool // Indicates whether v3io tsdb already queried and aggregated the data, or just returned raw data
isAlreadyV3IOAggregated func(op string) bool // Indicates whether v3io tsdb already queried and aggregated the data, or just returned raw data
}

// errorf causes a panic with the input formatted into an error.
Expand Down Expand Up @@ -980,7 +997,7 @@ func (ev *evaluator) eval(expr Expr) Value {

switch e := expr.(type) {
case *AggregateExpr:
if ev.useV3ioAggregations {
if ev.isAlreadyV3IOAggregated(e.Op.String()) {
return ev.emptyAggregation(e.Expr)
}
if s, ok := e.Param.(*StringLiteral); ok {
Expand Down Expand Up @@ -1038,9 +1055,10 @@ func (ev *evaluator) eval(expr Expr) Value {
// Evaluate any non-matrix arguments.
otherArgs := make([]Matrix, len(e.Args))
otherInArgs := make([]Vector, len(e.Args))
function := e.Func.Name
for i, e := range e.Args {
if i != matrixArgIndex {
if ev.useV3ioAggregations {
if ev.isAlreadyV3IOAggregated(function) {
return ev.emptyAggregation(e)
}

Expand Down Expand Up @@ -1226,14 +1244,14 @@ func (ev *evaluator) eval(expr Expr) Value {
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,
useV3ioAggregations: ev.useV3ioAggregations,
endTimestamp: ev.endTimestamp - offsetMillis,
interval: ev.defaultEvalInterval,
ctx: ev.ctx,
currentSamples: ev.currentSamples,
maxSamples: ev.maxSamples,
defaultEvalInterval: ev.defaultEvalInterval,
logger: ev.logger,
isAlreadyV3IOAggregated: ev.isAlreadyV3IOAggregated,
}

if e.Step != 0 {
Expand Down Expand Up @@ -1996,28 +2014,34 @@ func isV3ioEligibleFunction(function string) bool {
return supportedV3ioFunctions[function]
}

func isV3ioEligibleQueryExpr(e Expr) bool {
switch expr := e.(type) {
func isV3ioEligibleQueryExpr(p []Node) bool {
if len(p) == 0 {
return true
}
switch n := p[len(p)-1].(type) {
case *AggregateExpr:
if !isV3ioEligibleAggregation(expr.Op) {
if !isV3ioEligibleAggregation(n.Op) {
return false
}
if expr.Without {
if n.Without {
return false
}
// Currently only supports non-nested functions.
// Not supported - avg(max_over_time(cpu[10m])), Supported - avg(cpu)
if e, ok := expr.Expr.(*Call); ok {
if e, ok := n.Expr.(*Call); ok {
if e.Func != nil {
return false
}
}
return true
case *Call:
return isV3ioEligibleFunction(expr.Func.Name)
return isV3ioEligibleFunction(n.Func.Name)
case *BinaryExpr:
// If we hit a binary expression we terminate since we only care about functions
// or aggregations over a single metric.
return false
}

return false
return isV3ioEligibleQueryExpr(p[:len(p)-1])
}

// btos returns 1 if b is true, 0 otherwise.
Expand Down
35 changes: 31 additions & 4 deletions storage/tsdb/promtsdb.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -75,14 +75,22 @@ type V3ioPromQuerier struct {
logger logger.Logger
mint, maxt int64

UseAggregatesConfig bool // Indicate whether or not to use v3io aggregations by default (passed from prometheus.yml)
UseAggregates bool // Indicate whether the current query is eligible for using v3io aggregations (should be set after creating a Querier instance)
UseAggregatesConfig bool // Indicate whether or not to use v3io aggregations by default (passed from prometheus.yml)
UseAggregates bool // Indicate whether the current query is eligible for using v3io aggregations (should be set after creating a Querier instance)
LastTSDBAggregatedAggr string
}

func (promQuery *V3ioPromQuerier) UseV3ioAggregations() bool {
return promQuery.UseAggregates && promQuery.UseAggregatesConfig
}

func (promQuery *V3ioPromQuerier) IsAlreadyAggregated(op string) bool {
if promQuery.UseV3ioAggregations() && promQuery.LastTSDBAggregatedAggr == op {
return true
}
return false
}

// Select returns a set of series that matches the given label matchers.
func (promQuery *V3ioPromQuerier) Select(params *storage.SelectParams, oms ...*labels.Matcher) (storage.SeriesSet, storage.Warnings, error) {
name, filter, function := match2filter(oms, promQuery.logger)
Expand All @@ -101,15 +109,23 @@ func (promQuery *V3ioPromQuerier) Select(params *storage.SelectParams, oms ...*l
promQuery.logger.Debug("SelectParams: %+v", params)
overTimeSuffix := "_over_time"

// Currently we do aggregations only for:
// 1. All Cross-series aggregation
// 2. Over-time aggregations where no Step or aggregationWindow was specified
// 3. Over-time aggregation for v3io-tsdb compatible aggregates
// 4. Down sampling - when only a step is provided
// Note: in addition to the above cases, we also take into consider the `UseAggregatesConfig` configuration and of
// course whether or not the requested aggregation is a valid v3io-tsdb aggregation
if params.Func != "" {
// only pass xx_over_time functions (just the xx part)
// TODO: support count/stdxx, require changes in Prometheus: promql/functions.go, not calc aggregate twice
if strings.HasSuffix(params.Func, overTimeSuffix) {
if promQuery.UseAggregates && promQuery.UseAggregatesConfig {
if promQuery.UseV3ioAggregations() {
function = strings.TrimSuffix(params.Func, overTimeSuffix)
} else {
f := params.Func[0:3]
if params.Step == 0 && (f == "min" || f == "max" || f == "sum" || f == "avg") {
if params.Step == 0 && params.AggregationWindow == 0 &&
(f == "min" || f == "max" || f == "sum" || f == "avg") {
function = f
} else {
noAggr = true
Expand All @@ -120,6 +136,10 @@ func (promQuery *V3ioPromQuerier) Select(params *storage.SelectParams, oms ...*l
}
}

if function != "" && !noAggr {
promQuery.LastTSDBAggregatedAggr = params.Func
}

selectParams := &pquerier.SelectParams{Name: name,
Functions: function,
Step: params.Step,
Expand All @@ -128,6 +148,8 @@ func (promQuery *V3ioPromQuerier) Select(params *storage.SelectParams, oms ...*l
To: promQuery.maxt,
AggregationWindow: params.AggregationWindow}

promQuery.logger.DebugWith("Going to query tsdb", "params", selectParams,
"UseAggregates", promQuery.UseAggregates, "UseAggregatesConfig", promQuery.UseAggregatesConfig)
set, err := promQuery.v3ioQuerier.SelectProm(selectParams, noAggr)
return &V3ioPromSeriesSet{s: set}, nil, err
}
Expand Down Expand Up @@ -276,6 +298,11 @@ func (ls Labels) GetKey() (string, string, uint64) {

}


func (ls Labels) HashWithName() uint64 {
return ls.lbls.Hash()
}

// create update expression
func (ls Labels) GetExpr() string {
var lblExprBuilder strings.Builder
Expand Down
Loading

0 comments on commit 7a632e7

Please sign in to comment.