apitech
/
loki
mirror of https://github.com/grafana/loki
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Rename evaluator and its method to emphasize it's a factory. (#10381)

Browse Source
pull/10399/head
Karsten Jeschkies 3 years ago committed by GitHub
parent e3ab2c22de
commit 3a4c0e2d47
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 71 additions and 71 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 28
      pkg/logql/downstream.go
  2. 24
      pkg/logql/engine.go
  3. 84
      pkg/logql/evaluator.go
  4. 6
      pkg/logql/step_evaluator.go

28
pkg/logql/downstream.go
Unescape View File

@ -204,7 +204,7 @@ type Downstreamer interface {
// DownstreamEvaluator is an evaluator which handles shard aware AST nodes
type DownstreamEvaluator struct {
Downstreamer
defaultEvaluator Evaluator
defaultEvaluator EvaluatorFactory
}
// Downstream runs queries and collects stats from the embedded Downstreamer
@ -253,10 +253,10 @@ func NewDownstreamEvaluator(downstreamer Downstreamer) *DownstreamEvaluator {
}
}
// StepEvaluator returns a StepEvaluator for a given SampleExpr
func (ev *DownstreamEvaluator) StepEvaluator(
// NewStepEvaluator returns a NewStepEvaluator for a given SampleExpr
func (ev *DownstreamEvaluator) NewStepEvaluator(
ctx context.Context,
nextEv SampleEvaluator,
nextEvFactory SampleEvaluatorFactory,
expr syntax.SampleExpr,
params Params,
) (StepEvaluator, error) {
@ -276,7 +276,7 @@ func (ev *DownstreamEvaluator) StepEvaluator(
if err != nil {
return nil, err
}
return ResultStepEvaluator(results[0], params)
return NewResultStepEvaluator(results[0], params)
case *ConcatSampleExpr:
cur := e
@ -300,7 +300,7 @@ func (ev *DownstreamEvaluator) StepEvaluator(
xs := make([]StepEvaluator, 0, len(queries))
for i, res := range results {
stepper, err := ResultStepEvaluator(res, params)
stepper, err := NewResultStepEvaluator(res, params)
if err != nil {
level.Warn(util_log.Logger).Log(
"msg", "could not extract StepEvaluator",
@ -312,15 +312,15 @@ func (ev *DownstreamEvaluator) StepEvaluator(
xs = append(xs, stepper)
}
return ConcatEvaluator(xs)
return NewConcatStepEvaluator(xs)
default:
return ev.defaultEvaluator.StepEvaluator(ctx, nextEv, e, params)
return ev.defaultEvaluator.NewStepEvaluator(ctx, nextEvFactory, e, params)
}
}
// Iterator returns the iter.EntryIterator for a given LogSelectorExpr
func (ev *DownstreamEvaluator) Iterator(
// NewIterator returns the iter.EntryIterator for a given LogSelectorExpr
func (ev *DownstreamEvaluator) NewIterator(
ctx context.Context,
expr syntax.LogSelectorExpr,
params Params,
@ -382,9 +382,9 @@ func (ev *DownstreamEvaluator) Iterator(
}
}
// ConcatEvaluator joins multiple StepEvaluators.
// NewConcatStepEvaluator joins multiple StepEvaluators.
// Contract: They must be of identical start, end, and step values.
func ConcatEvaluator(evaluators []StepEvaluator) (StepEvaluator, error) {
func NewConcatStepEvaluator(evaluators []StepEvaluator) (StepEvaluator, error) {
return newStepEvaluator(
func() (ok bool, ts int64, vec promql.Vector) {
var cur promql.Vector
@ -421,8 +421,8 @@ func ConcatEvaluator(evaluators []StepEvaluator) (StepEvaluator, error) {
)
}
// ResultStepEvaluator coerces a downstream vector or matrix into a StepEvaluator
func ResultStepEvaluator(res logqlmodel.Result, params Params) (StepEvaluator, error) {
// NewResultStepEvaluator coerces a downstream vector or matrix into a StepEvaluator
func NewResultStepEvaluator(res logqlmodel.Result, params Params) (StepEvaluator, error) {
var (
start = params.Start()
end = params.End()

24
pkg/logql/engine.go
Unescape View File

@ -135,10 +135,10 @@ func (opts *EngineOpts) applyDefault() {
// Engine is the LogQL engine.
type Engine struct {
logger log.Logger
evaluator Evaluator
limits Limits
opts EngineOpts
logger log.Logger
evaluatorFactory EvaluatorFactory
limits Limits
opts EngineOpts
}
// NewEngine creates a new LogQL Engine.
@ -148,10 +148,10 @@ func NewEngine(opts EngineOpts, q Querier, l Limits, logger log.Logger) *Engine
logger = log.NewNopLogger()
}
return &Engine{
logger: logger,
evaluator: NewDefaultEvaluator(q, opts.MaxLookBackPeriod),
limits: l,
opts: opts,
logger: logger,
evaluatorFactory: NewDefaultEvaluator(q, opts.MaxLookBackPeriod),
limits: l,
opts: opts,
}
}
@ -160,7 +160,7 @@ func (ng *Engine) Query(params Params) Query {
return &query{
logger: ng.logger,
params: params,
evaluator: ng.evaluator,
evaluator: ng.evaluatorFactory,
parse: func(_ context.Context, query string) (syntax.Expr, error) {
return syntax.ParseExpr(query)
},
@ -181,7 +181,7 @@ type query struct {
params Params
parse func(context.Context, string) (syntax.Expr, error)
limits Limits
evaluator Evaluator
evaluator EvaluatorFactory
record bool
logExecQuery bool
}
@ -286,7 +286,7 @@ func (q *query) Eval(ctx context.Context) (promql_parser.Value, error) {
return value, err
case syntax.LogSelectorExpr:
iter, err := q.evaluator.Iterator(ctx, e, q.params)
iter, err := q.evaluator.NewIterator(ctx, e, q.params)
if err != nil {
return nil, err
}
@ -340,7 +340,7 @@ func (q *query) evalSample(ctx context.Context, expr syntax.SampleExpr) (promql_
return nil, err
}
stepEvaluator, err := q.evaluator.StepEvaluator(ctx, q.evaluator, expr, q.params)
stepEvaluator, err := q.evaluator.NewStepEvaluator(ctx, q.evaluator, expr, q.params)
if err != nil {
return nil, err
}

84
pkg/logql/evaluator.go
Unescape View File

@ -123,31 +123,31 @@ func Sortable(q Params) (bool, error) {
return sortable, nil
}
// Evaluator is an interface for iterating over data at different nodes in the AST
type Evaluator interface {
SampleEvaluator
EntryEvaluator
// EvaluatorFactory is an interface for iterating over data at different nodes in the AST
type EvaluatorFactory interface {
SampleEvaluatorFactory
EntryEvaluatorFactory
}
type SampleEvaluator interface {
// StepEvaluator returns a StepEvaluator for a given SampleExpr. It's explicitly passed another StepEvaluator// in order to enable arbitrary computation of embedded expressions. This allows more modular & extensible
// StepEvaluator implementations which can be composed.
StepEvaluator(ctx context.Context, nextEvaluator SampleEvaluator, expr syntax.SampleExpr, p Params) (StepEvaluator, error)
type SampleEvaluatorFactory interface {
// NewStepEvaluator returns a NewStepEvaluator for a given SampleExpr. It's explicitly passed another NewStepEvaluator// in order to enable arbitrary computation of embedded expressions. This allows more modular & extensible
// NewStepEvaluator implementations which can be composed.
NewStepEvaluator(ctx context.Context, nextEvaluatorFactory SampleEvaluatorFactory, expr syntax.SampleExpr, p Params) (StepEvaluator, error)
}
type SampleEvaluatorFunc func(ctx context.Context, nextEvaluator SampleEvaluator, expr syntax.SampleExpr, p Params) (StepEvaluator, error)
type SampleEvaluatorFunc func(ctx context.Context, nextEvaluatorFactory SampleEvaluatorFactory, expr syntax.SampleExpr, p Params) (StepEvaluator, error)
func (s SampleEvaluatorFunc) StepEvaluator(ctx context.Context, nextEvaluator SampleEvaluator, expr syntax.SampleExpr, p Params) (StepEvaluator, error) {
return s(ctx, nextEvaluator, expr, p)
func (s SampleEvaluatorFunc) NewStepEvaluator(ctx context.Context, nextEvaluatorFactory SampleEvaluatorFactory, expr syntax.SampleExpr, p Params) (StepEvaluator, error) {
return s(ctx, nextEvaluatorFactory, expr, p)
}
type EntryEvaluator interface {
// Iterator returns the iter.EntryIterator for a given LogSelectorExpr
Iterator(context.Context, syntax.LogSelectorExpr, Params) (iter.EntryIterator, error)
type EntryEvaluatorFactory interface {
// NewIterator returns the iter.EntryIterator for a given LogSelectorExpr
NewIterator(context.Context, syntax.LogSelectorExpr, Params) (iter.EntryIterator, error)
}
// EvaluatorUnsupportedType is a helper for signaling that an evaluator does not support an Expr type
func EvaluatorUnsupportedType(expr syntax.Expr, ev Evaluator) error {
func EvaluatorUnsupportedType(expr syntax.Expr, ev EvaluatorFactory) error {
return errors.Errorf("unexpected expr type (%T) for Evaluator type (%T) ", expr, ev)
}
@ -164,7 +164,7 @@ func NewDefaultEvaluator(querier Querier, maxLookBackPeriod time.Duration) *Defa
}
}
func (ev *DefaultEvaluator) Iterator(ctx context.Context, expr syntax.LogSelectorExpr, q Params) (iter.EntryIterator, error) {
func (ev *DefaultEvaluator) NewIterator(ctx context.Context, expr syntax.LogSelectorExpr, q Params) (iter.EntryIterator, error) {
params := SelectLogParams{
QueryRequest: &logproto.QueryRequest{
Start: q.Start(),
@ -183,9 +183,9 @@ func (ev *DefaultEvaluator) Iterator(ctx context.Context, expr syntax.LogSelecto
return ev.querier.SelectLogs(ctx, params)
}
func (ev *DefaultEvaluator) StepEvaluator(
func (ev *DefaultEvaluator) NewStepEvaluator(
ctx context.Context,
nextEv SampleEvaluator,
nextEvFactory SampleEvaluatorFactory,
expr syntax.SampleExpr,
q Params,
) (StepEvaluator, error) {
@ -194,7 +194,7 @@ func (ev *DefaultEvaluator) StepEvaluator(
if rangExpr, ok := e.Left.(*syntax.RangeAggregationExpr); ok && e.Operation == syntax.OpTypeSum {
// if range expression is wrapped with a vector expression
// we should send the vector expression for allowing reducing labels at the source.
nextEv = SampleEvaluatorFunc(func(ctx context.Context, _ SampleEvaluator, _ syntax.SampleExpr, _ Params) (StepEvaluator, error) {
nextEvFactory = SampleEvaluatorFunc(func(ctx context.Context, _ SampleEvaluatorFactory, _ syntax.SampleExpr, _ Params) (StepEvaluator, error) {
it, err := ev.querier.SelectSamples(ctx, SelectSampleParams{
&logproto.SampleQueryRequest{
Start: q.Start().Add(-rangExpr.Left.Interval).Add(-rangExpr.Left.Offset),
@ -206,10 +206,10 @@ func (ev *DefaultEvaluator) StepEvaluator(
if err != nil {
return nil, err
}
return rangeAggEvaluator(iter.NewPeekingSampleIterator(it), rangExpr, q, rangExpr.Left.Offset)
return newRangeAggEvaluator(iter.NewPeekingSampleIterator(it), rangExpr, q, rangExpr.Left.Offset)
})
}
return vectorAggEvaluator(ctx, nextEv, e, q)
return newVectorAggEvaluator(ctx, nextEvFactory, e, q)
case *syntax.RangeAggregationExpr:
it, err := ev.querier.SelectSamples(ctx, SelectSampleParams{
&logproto.SampleQueryRequest{
@ -222,11 +222,11 @@ func (ev *DefaultEvaluator) StepEvaluator(
if err != nil {
return nil, err
}
return rangeAggEvaluator(iter.NewPeekingSampleIterator(it), e, q, e.Left.Offset)
return newRangeAggEvaluator(iter.NewPeekingSampleIterator(it), e, q, e.Left.Offset)
case *syntax.BinOpExpr:
return binOpStepEvaluator(ctx, nextEv, e, q)
return newBinOpStepEvaluator(ctx, nextEvFactory, e, q)
case *syntax.LabelReplaceExpr:
return labelReplaceEvaluator(ctx, nextEv, e, q)
return newLabelReplaceEvaluator(ctx, nextEvFactory, e, q)
case *syntax.VectorExpr:
val, err := e.Value()
if err != nil {
@ -238,16 +238,16 @@ func (ev *DefaultEvaluator) StepEvaluator(
}
}
func vectorAggEvaluator(
func newVectorAggEvaluator(
ctx context.Context,
ev SampleEvaluator,
evFactory SampleEvaluatorFactory,
expr *syntax.VectorAggregationExpr,
q Params,
) (StepEvaluator, error) {
if expr.Grouping == nil {
return nil, errors.Errorf("aggregation operator '%q' without grouping", expr.Operation)
}
nextEvaluator, err := ev.StepEvaluator(ctx, ev, expr.Left, q)
nextEvaluator, err := evFactory.NewStepEvaluator(ctx, evFactory, expr.Left, q)
if err != nil {
return nil, err
}
@ -450,7 +450,7 @@ func vectorAggEvaluator(
}, nextEvaluator.Close, nextEvaluator.Error)
}
func rangeAggEvaluator(
func newRangeAggEvaluator(
it iter.PeekingSampleIterator,
expr *syntax.RangeAggregationExpr,
q Params,
@ -555,9 +555,9 @@ func (r absentRangeVectorEvaluator) Error() error {
// binOpExpr explicitly does not handle when both legs are literals as
// it makes the type system simpler and these are reduced in mustNewBinOpExpr
func binOpStepEvaluator(
func newBinOpStepEvaluator(
ctx context.Context,
ev SampleEvaluator,
evFactory SampleEvaluatorFactory,
expr *syntax.BinOpExpr,
q Params,
) (StepEvaluator, error) {
@ -567,11 +567,11 @@ func binOpStepEvaluator(
// match a literal expr with all labels in the other leg
if lOk {
rhs, err := ev.StepEvaluator(ctx, ev, expr.RHS, q)
rhs, err := evFactory.NewStepEvaluator(ctx, evFactory, expr.RHS, q)
if err != nil {
return nil, err
}
return literalStepEvaluator(
return newLiteralStepEvaluator(
expr.Op,
leftLit,
rhs,
@ -580,11 +580,11 @@ func binOpStepEvaluator(
)
}
if rOk {
lhs, err := ev.StepEvaluator(ctx, ev, expr.SampleExpr, q)
lhs, err := evFactory.NewStepEvaluator(ctx, evFactory, expr.SampleExpr, q)
if err != nil {
return nil, err
}
return literalStepEvaluator(
return newLiteralStepEvaluator(
expr.Op,
rightLit,
lhs,
@ -602,7 +602,7 @@ func binOpStepEvaluator(
// load them in parallel
g.Go(func() error {
var err error
lse, err = ev.StepEvaluator(ctx, ev, expr.SampleExpr, q)
lse, err = evFactory.NewStepEvaluator(ctx, evFactory, expr.SampleExpr, q)
if err != nil {
cancel()
}
@ -610,7 +610,7 @@ func binOpStepEvaluator(
})
g.Go(func() error {
var err error
rse, err = ev.StepEvaluator(ctx, ev, expr.RHS, q)
rse, err = evFactory.NewStepEvaluator(ctx, evFactory, expr.RHS, q)
if err != nil {
cancel()
}
@ -877,9 +877,9 @@ func resultMetric(lhs, rhs labels.Labels, opts *syntax.BinOpOptions) labels.Labe
return lb.Labels()
}
// literalStepEvaluator merges a literal with a StepEvaluator. Since order matters in
// newLiteralStepEvaluator merges a literal with a StepEvaluator. Since order matters in
// non-commutative operations, inverted should be true when the literalExpr is not the left argument.
func literalStepEvaluator(
func newLiteralStepEvaluator(
op string,
lit *syntax.LiteralExpr,
eval StepEvaluator,
@ -974,14 +974,14 @@ func (r *vectorIterator) Error() error {
return nil
}
// labelReplaceEvaluator
func labelReplaceEvaluator(
// newLabelReplaceEvaluator
func newLabelReplaceEvaluator(
ctx context.Context,
ev SampleEvaluator,
evFactory SampleEvaluatorFactory,
expr *syntax.LabelReplaceExpr,
q Params,
) (StepEvaluator, error) {
nextEvaluator, err := ev.StepEvaluator(ctx, ev, expr.Left, q)
nextEvaluator, err := evFactory.NewStepEvaluator(ctx, evFactory, expr.Left, q)
if err != nil {
return nil, err
}

6
pkg/logql/step_evaluator.go
Unescape View File

@ -17,7 +17,7 @@ type StepEvaluator interface {
}
type stepEvaluator struct {
fn func() (bool, int64, promql.Vector)
next func() (bool, int64, promql.Vector)
close func() error
err func() error
}
@ -35,14 +35,14 @@ func newStepEvaluator(fn func() (bool, int64, promql.Vector), closeFn func() err
err = func() error { return nil }
}
return &stepEvaluator{
fn: fn,
next: fn,
close: closeFn,
err: err,
}, nil
}
func (e *stepEvaluator) Next() (bool, int64, promql.Vector) {
return e.fn()
return e.next()
}
func (e *stepEvaluator) Close() error {

Write Preview
Loading…
Cancel
Save