Improve pipeline stages ast errors.

pkg/logql/ast.go

@@ -119,7 +119,7 @@ func (m MultiStageExpr) stages() ([]log.Stage, error) {
 	for _, e := range m {
 		p, err := e.Stage()
 		if err != nil {
-			return nil, err
+			return nil, newStageError(e, err)
 		}
 		if p == log.NoopStage {
 			continue

pkg/logql/error.go

@@ -0,0 +1,70 @@
+package logql
+
+import (
+	"fmt"
+
+	"github.com/grafana/loki/pkg/logql/log"
+	"github.com/prometheus/prometheus/pkg/labels"
+)
+
+// ParseError is what is returned when we failed to parse.
+type ParseError struct {
+	msg       string
+	line, col int
+}
+
+func (p ParseError) Error() string {
+	if p.col == 0 && p.line == 0 {
+		return fmt.Sprintf("parse error : %s", p.msg)
+	}
+	return fmt.Sprintf("parse error at line %d, col %d: %s", p.line, p.col, p.msg)
+}
+
+func newParseError(msg string, line, col int) ParseError {
+	return ParseError{
+		msg:  msg,
+		line: line,
+		col:  col,
+	}
+}
+
+func newStageError(expr Expr, err error) ParseError {
+	return ParseError{
+		msg:  fmt.Sprintf(`stage '%s' : %s`, expr, err),
+		line: 0,
+		col:  0,
+	}
+}
+
+// IsParseError returns true if the err is a ast parsing error.
+func IsParseError(err error) bool {
+	_, ok := err.(ParseError)
+	return ok
+}
+
+type pipelineError struct {
+	metric    labels.Labels
+	errorType string
+}
+
+func newPipelineErr(metric labels.Labels) *pipelineError {
+	return &pipelineError{
+		metric:    metric,
+		errorType: metric.Get(log.ErrorLabel),
+	}
+}
+
+func (e pipelineError) Error() string {
+	return fmt.Sprintf(
+		"pipeline error: '%s' for series: '%s'.\n"+
+			"Use a label filter to intentionally skip this error. (e.g | __error__!=\"%s\").\n"+
+			"To skip all potential errors you can match empty errors.(e.g __error__=\"\")\n"+
+			"The label filter can also be specified after unwrap. (e.g | unwrap latency | __error__=\"\" )\n",
+		e.errorType, e.metric, e.errorType)
+}
+
+// IsPipelineError tells if the error is generated by a Pipeline.
+func IsPipelineError(err error) bool {
+	_, ok := err.(*pipelineError)
+	return ok
+}

pkg/logql/evaluator.go

@@ -435,33 +435,6 @@ type rangeVectorEvaluator struct {
 	err error
 }
 
-type pipelineError struct {
-	metric    labels.Labels
-	errorType string
-}
-
-func newPipelineErr(metric labels.Labels) *pipelineError {
-	return &pipelineError{
-		metric:    metric,
-		errorType: metric.Get(log.ErrorLabel),
-	}
-}
-
-func (e pipelineError) Error() string {
-	return fmt.Sprintf(
-		"pipeline error: '%s' for series: '%s'.\n"+
-			"Use a label filter to intentionally skip this error. (e.g | __error__!=\"%s\").\n"+
-			"To skip all potential errors you can match empty errors.(e.g __error__=\"\")\n"+
-			"The label filter can also be specified after unwrap. (e.g | unwrap latency | __error__=\"\" )\n",
-		e.errorType, e.metric, e.errorType)
-}
-
-// IsPipelineError tells if the error is generated by a Pipeline.
-func IsPipelineError(err error) bool {
-	_, ok := err.(*pipelineError)
-	return ok
-}
-
 func (r *rangeVectorEvaluator) Next() (bool, int64, promql.Vector) {
 	next := r.iter.Next()
 	if !next {

pkg/logql/parser.go

@@ -2,7 +2,6 @@ package logql
 
 import (
 	"errors"
-	"fmt"
 	"strings"
 	"text/scanner"
 
@@ -86,30 +85,3 @@ func ParseLogSelector(input string) (LogSelectorExpr, error) {
 	}
 	return logSelector, nil
 }
-
-// ParseError is what is returned when we failed to parse.
-type ParseError struct {
-	msg       string
-	line, col int
-}
-
-func (p ParseError) Error() string {
-	if p.col == 0 && p.line == 0 {
-		return fmt.Sprintf("parse error : %s", p.msg)
-	}
-	return fmt.Sprintf("parse error at line %d, col %d: %s", p.line, p.col, p.msg)
-}
-
-func newParseError(msg string, line, col int) ParseError {
-	return ParseError{
-		msg:  msg,
-		line: line,
-		col:  col,
-	}
-}
-
-// IsParseError returns true if the err is a ast parsing error.
-func IsParseError(err error) bool {
-	_, ok := err.(ParseError)
-	return ok
-}