Refactoring in progress.

Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com>

pkg/logql/ast.go

@@ -14,6 +14,7 @@ import (
 	"github.com/grafana/loki/pkg/iter"
 	"github.com/grafana/loki/pkg/logproto"
 	"github.com/grafana/loki/pkg/logql/labelfilter"
+	"github.com/grafana/loki/pkg/logql/log"
 )
 
 // Expr is the root expression which can be a SampleExpr or LogSelectorExpr
@@ -81,62 +82,36 @@ type LogSelectorExpr interface {
 }
 
 type PipelineExpr interface {
-	Pipeline() (Pipeline, error)
+	Pipeline() (log.Pipeline, error)
 	Expr
 }
 
-type Pipeline interface {
-	Process(line []byte, lbs labels.Labels) ([]byte, labels.Labels, bool)
-}
-
-var NoopPipeline = &noopPipeline{}
-
-type noopPipeline struct{}
-
-func (noopPipeline) Process(line []byte, lbs labels.Labels) ([]byte, labels.Labels, bool) {
-	return line, lbs, true
-}
-
-type PipelineFunc func(line []byte, lbs labels.Labels) ([]byte, labels.Labels, bool)
-
-func (fn PipelineFunc) Process(line []byte, lbs labels.Labels) ([]byte, labels.Labels, bool) {
-	return fn(line, lbs)
-}
-
-type MultiPipeline []Pipeline
-
-func (m MultiPipeline) Process(line []byte, lbs labels.Labels) ([]byte, labels.Labels, bool) {
-	var ok bool
-	for _, p := range m {
-		line, lbs, ok = p.Process(line, lbs)
-		if !ok {
-			return line, lbs, ok
-		}
-	}
-	return line, lbs, ok
+type StageExpr interface {
+	Stage() (log.Stage, error)
+	Expr
 }
 
-type MultiPipelineExpr []PipelineExpr
+type MultiStageExpr []StageExpr
 
-func (m MultiPipelineExpr) Pipeline() (Pipeline, error) {
-	c := make(MultiPipeline, 0, len(m))
+func (m MultiStageExpr) Pipeline() (log.Pipeline, error) {
+	c := make(log.MultiStage, 0, len(m))
 	for _, e := range m {
-		p, err := e.Pipeline()
+		p, err := e.Stage()
 		if err != nil {
 			return nil, err
 		}
-		if p == NoopPipeline {
+		if p == log.NoopStage {
 			continue
 		}
 		c = append(c, p)
 	}
 	if len(c) == 0 {
-		return NoopPipeline, nil
+		return log.NoopPipeline, nil
 	}
 	return c, nil
 }
 
-func (m MultiPipelineExpr) String() string {
+func (m MultiStageExpr) String() string {
 	var sb strings.Builder
 	for i, e := range m {
 		sb.WriteString(e.String())
@@ -147,7 +122,7 @@ func (m MultiPipelineExpr) String() string {
 	return sb.String()
 }
 
-func (MultiPipelineExpr) logQLExpr() {}
+func (MultiStageExpr) logQLExpr() {}
 
 func FilterToPipeline(f LineFilter) Pipeline {
 	if f == nil || f == TrueFilter {
@@ -191,21 +166,21 @@ func (e *matchersExpr) String() string {
 	return sb.String()
 }
 
-func (e *matchersExpr) Pipeline() (Pipeline, error) {
-	return NoopPipeline, nil
+func (e *matchersExpr) Pipeline() (log.Pipeline, error) {
+	return log.NoopPipeline, nil
 }
 
 func (e *matchersExpr) HasFilter() bool {
 	return false
 }
 
 type pipelineExpr struct {
-	pipeline MultiPipelineExpr
+	pipeline MultiStageExpr
 	left     *matchersExpr
 	implicit
 }
 
-func newPipelineExpr(left *matchersExpr, pipeline MultiPipelineExpr) LogSelectorExpr {
+func newPipelineExpr(left *matchersExpr, pipeline MultiStageExpr) LogSelectorExpr {
 	return &pipelineExpr{
 		left:     left,
 		pipeline: pipeline,
@@ -224,7 +199,7 @@ func (e *pipelineExpr) String() string {
 	return sb.String()
 }
 
-func (e *pipelineExpr) Pipeline() (Pipeline, error) {
+func (e *pipelineExpr) Pipeline() (log.Pipeline, error) {
 	return e.pipeline.Pipeline()
 }
 
@@ -291,8 +266,8 @@ func (e *lineFilterExpr) String() string {
 	return sb.String()
 }
 
-func (e *lineFilterExpr) Filter() (LineFilter, error) {
-	f, err := newFilter(e.match, e.ty)
+func (e *lineFilterExpr) Filter() (log.Filterer, error) {
+	f, err := log.NewFilter(e.match, e.ty)
 	if err != nil {
 		return nil, err
 	}
@@ -302,11 +277,11 @@ func (e *lineFilterExpr) Filter() (LineFilter, error) {
 			return nil, err
 		}
 		if nextFilter != nil {
-			f = newAndFilter(nextFilter, f)
+			f = log.NewAndFilter(nextFilter, f)
 		}
 	}
 
-	if f == TrueFilter {
+	if f == log.TrueFilter {
 		return nil, nil
 	}
 

pkg/logql/expr.y

@@ -33,8 +33,8 @@ import (
   BinOpModifier           BinOpOptions
   LabelParser             *labelParserExpr
   LineFilters             *lineFilterExpr
-  PipelineExpr            MultiPipelineExpr
-  PipelineStage           PipelineExpr
+  PipelineExpr            MultiStageExpr
+  PipelineStage           StageExpr
   BytesFilter             labelfilter.Filterer
   NumberFilter            labelfilter.Filterer
   DurationFilter          labelfilter.Filterer
@@ -192,7 +192,7 @@ matcher:
     ;
 
 pipelineExpr:
-      pipelineStage                  { $$ = MultiPipelineExpr{ $1 } }
+      pipelineStage                  { $$ = MultiStageExpr{ $1 } }
     | pipelineExpr pipelineStage     { $$ = append($1, $2)}
     ;
 

pkg/logql/labelfilter/bytes.go

@@ -9,11 +9,11 @@ import (
 
 type Bytes struct {
 	Name  string
-	Value uint64 
+	Value uint64
 	Type  FilterType
 }
 
-func NewBytes(t FilterType, name string, b uint64) *Bytes{
+func NewBytes(t FilterType, name string, b uint64) *Bytes {
 	return &Bytes{
 		Name:  name,
 		Type:  t,

pkg/logql/error.go → pkg/logql/log/error.go

@@ -1,4 +1,4 @@
-package logql
+package log
 
 var (
 	errJSON             = "JSONParser"

pkg/logql/filter.go → pkg/logql/log/filter.go

@@ -1,4 +1,4 @@
-package logql
+package log
 
 import (
 	"bytes"
@@ -9,15 +9,15 @@ import (
 	"github.com/prometheus/prometheus/pkg/labels"
 )
 
-// LineFilter is a interface to filter log lines.
-type LineFilter interface {
+// Filterer is a interface to filter log lines.
+type Filterer interface {
 	Filter(line []byte) bool
 }
 
 // LineFilterFunc is a syntax sugar for creating line filter from a function
-type LineFilterFunc func(line []byte) bool
+type FiltererFunc func(line []byte) bool
 
-func (f LineFilterFunc) Filter(line []byte) bool {
+func (f FiltererFunc) Filter(line []byte) bool {
 	return f(line)
 }
 
@@ -29,30 +29,30 @@ func (trueFilter) Filter(_ []byte) bool { return true }
 var TrueFilter = trueFilter{}
 
 type notFilter struct {
-	LineFilter
+	Filterer
 }
 
 func (n notFilter) Filter(line []byte) bool {
-	return !n.LineFilter.Filter(line)
+	return !n.Filterer.Filter(line)
 }
 
 // newNotFilter creates a new filter which matches only if the base filter doesn't match.
 // If the base filter is a `or` it will recursively simplify with `and` operations.
-func newNotFilter(base LineFilter) LineFilter {
+func newNotFilter(base Filterer) Filterer {
 	// not(a|b) = not(a) and not(b) , and operation can't benefit from this optimization because both legs always needs to be executed.
 	if or, ok := base.(orFilter); ok {
-		return newAndFilter(newNotFilter(or.left), newNotFilter(or.right))
+		return NewAndFilter(newNotFilter(or.left), newNotFilter(or.right))
 	}
-	return notFilter{LineFilter: base}
+	return notFilter{Filterer: base}
 }
 
 type andFilter struct {
-	left  LineFilter
-	right LineFilter
+	left  Filterer
+	right Filterer
 }
 
-// newAndFilter creates a new filter which matches only if left and right matches.
-func newAndFilter(left LineFilter, right LineFilter) LineFilter {
+// NewAndFilter creates a new filter which matches only if left and right matches.
+func NewAndFilter(left Filterer, right Filterer) Filterer {
 	// Make sure we take care of panics in case a nil or noop filter is passed.
 	if right == nil || right == TrueFilter {
 		return left
@@ -73,12 +73,12 @@ func (a andFilter) Filter(line []byte) bool {
 }
 
 type orFilter struct {
-	left  LineFilter
-	right LineFilter
+	left  Filterer
+	right Filterer
 }
 
 // newOrFilter creates a new filter which matches only if left or right matches.
-func newOrFilter(left LineFilter, right LineFilter) LineFilter {
+func newOrFilter(left Filterer, right Filterer) Filterer {
 	if left == nil || left == TrueFilter {
 		return right
 	}
@@ -94,7 +94,7 @@ func newOrFilter(left LineFilter, right LineFilter) LineFilter {
 }
 
 // chainOrFilter is a syntax sugar to chain multiple `or` filters. (1 or many)
-func chainOrFilter(curr, new LineFilter) LineFilter {
+func chainOrFilter(curr, new Filterer) Filterer {
 	if curr == nil {
 		return new
 	}
@@ -111,7 +111,7 @@ type regexpFilter struct {
 
 // newRegexpFilter creates a new line filter for a given regexp.
 // If match is false the filter is the negation of the regexp.
-func newRegexpFilter(re string, match bool) (LineFilter, error) {
+func newRegexpFilter(re string, match bool) (Filterer, error) {
 	reg, err := regexp.Compile(re)
 	if err != nil {
 		return nil, err
@@ -143,7 +143,7 @@ func (l containsFilter) String() string {
 	return string(l.match)
 }
 
-func newContainsFilter(match []byte, caseInsensitive bool) LineFilter {
+func newContainsFilter(match []byte, caseInsensitive bool) Filterer {
 	if len(match) == 0 {
 		return TrueFilter
 	}
@@ -156,8 +156,8 @@ func newContainsFilter(match []byte, caseInsensitive bool) LineFilter {
 	}
 }
 
-// newFilter creates a new line filter from a match string and type.
-func newFilter(match string, mt labels.MatchType) (LineFilter, error) {
+// NewFilter creates a new line filter from a match string and type.
+func NewFilter(match string, mt labels.MatchType) (Filterer, error) {
 	switch mt {
 	case labels.MatchRegexp:
 		return parseRegexpFilter(match, true)
@@ -174,7 +174,7 @@ func newFilter(match string, mt labels.MatchType) (LineFilter, error) {
 
 // parseRegexpFilter parses a regexp and attempt to simplify it with only literal filters.
 // If not possible it will returns the original regexp filter.
-func parseRegexpFilter(re string, match bool) (LineFilter, error) {
+func parseRegexpFilter(re string, match bool) (Filterer, error) {
 	reg, err := syntax.Parse(re, syntax.Perl)
 	if err != nil {
 		return nil, err
@@ -194,7 +194,7 @@ func parseRegexpFilter(re string, match bool) (LineFilter, error) {
 
 // simplify a regexp expression by replacing it, when possible, with a succession of literal filters.
 // For example `(foo|bar)` will be replaced by  `containsFilter(foo) or containsFilter(bar)`
-func simplify(reg *syntax.Regexp) (LineFilter, bool) {
+func simplify(reg *syntax.Regexp) (Filterer, bool) {
 	switch reg.Op {
 	case syntax.OpAlternate:
 		return simplifyAlternate(reg)
@@ -230,7 +230,7 @@ func clearCapture(regs ...*syntax.Regexp) {
 
 // simplifyAlternate simplifies, when possible, alternate regexp expressions such as:
 // (foo|bar) or (foo|(bar|buzz)).
-func simplifyAlternate(reg *syntax.Regexp) (LineFilter, bool) {
+func simplifyAlternate(reg *syntax.Regexp) (Filterer, bool) {
 	clearCapture(reg.Sub...)
 	// attempt to simplify the first leg
 	f, ok := simplify(reg.Sub[0])
@@ -253,15 +253,15 @@ func simplifyAlternate(reg *syntax.Regexp) (LineFilter, bool) {
 // which is a literalFilter.
 // Or a literal and alternates operation (see simplifyConcatAlternate), which represent a multiplication of alternates.
 // Anything else is rejected.
-func simplifyConcat(reg *syntax.Regexp, baseLiteral []byte) (LineFilter, bool) {
+func simplifyConcat(reg *syntax.Regexp, baseLiteral []byte) (Filterer, bool) {
 	clearCapture(reg.Sub...)
 	// we support only simplication of concat operation with 3 sub expressions.
 	// for instance .*foo.*bar contains 4 subs (.*+foo+.*+bar) and can't be simplified.
 	if len(reg.Sub) > 3 {
 		return nil, false
 	}
 
-	var curr LineFilter
+	var curr Filterer
 	var ok bool
 	literals := 0
 	for _, sub := range reg.Sub {
@@ -304,7 +304,7 @@ func simplifyConcat(reg *syntax.Regexp, baseLiteral []byte) (LineFilter, bool) {
 // A concat alternate is found when a concat operation has a sub alternate and is preceded by a literal.
 // For instance bar|b|buzz is expressed as b(ar|(?:)|uzz) => b concat alternate(ar,(?:),uzz).
 // (?:) being an OpEmptyMatch and b being the literal to concat all alternates (ar,(?:),uzz) with.
-func simplifyConcatAlternate(reg *syntax.Regexp, literal []byte, curr LineFilter) (LineFilter, bool) {
+func simplifyConcatAlternate(reg *syntax.Regexp, literal []byte, curr Filterer) (Filterer, bool) {
 	for _, alt := range reg.Sub {
 		switch alt.Op {
 		case syntax.OpEmptyMatch: