Add more tests and fix all engine tests

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

pkg/logql/engine_test.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"math"
+	"strings"
 
 	// "math"
 	"testing"
@@ -1817,7 +1818,7 @@ func paramsID(p interface{}) string {
 	if err != nil {
 		panic(err)
 	}
-	return string(b)
+	return strings.ReplaceAll(string(b), " ", "")
 }
 
 type logData struct {

pkg/logql/series_extractor_test.go

@@ -0,0 +1,38 @@
+package logql
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/prometheus/prometheus/pkg/labels"
+)
+
+func Test_labelSampleExtractor_Extract(t *testing.T) {
+	tests := []struct {
+		name    string
+		ex      *labelSampleExtractor
+		in      labels.Labels
+		want    float64
+		wantLbs labels.Labels
+	}{
+		{
+			"convert float",
+			newLabelSampleExtractor("foo"),
+			labels.Labels{labels.Label{Name: "foo", Value: "15.0"}},
+			15,
+			labels.Labels{},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+
+			outval, outlbs := tt.ex.Extract([]byte(""), tt.in)
+			if outval != tt.want {
+				t.Errorf("labelSampleExtractor.Extract() val = %v, want %v", outval, tt.want)
+			}
+			if !reflect.DeepEqual(outlbs, tt.wantLbs) {
+				t.Errorf("labelSampleExtractor.Extract() lbs = %v, want %v", outlbs, tt.wantLbs)
+			}
+		})
+	}
+}

pkg/logql/shardmapper.go

@@ -130,8 +130,8 @@ func (m ShardMapper) Map(expr Expr, r *shardRecorder) (Expr, error) {
 	case *literalExpr:
 		return e, nil
 	//todo(cyriltovena) enable sharding on logqlv2
-	// case *matchersExpr, *filterExpr:
-	// 	return m.mapLogSelectorExpr(e.(LogSelectorExpr), r), nil
+	case *matchersExpr, *pipelineExpr:
+		return m.mapLogSelectorExpr(e.(LogSelectorExpr), r), nil
 	case *vectorAggregationExpr:
 		return m.mapVectorAggregationExpr(e, r)
 	case *rangeAggregationExpr:

pkg/logql/shardmapper_test.go

@@ -1,6 +1,7 @@
 package logql
 
 import (
+	"strings"
 	"testing"
 	"time"
 
@@ -130,6 +131,10 @@ func TestMappingStrings(t *testing.T) {
 			in:  `{foo="bar"}`,
 			out: `downstream<{foo="bar"}, shard=0_of_2> ++ downstream<{foo="bar"}, shard=1_of_2>`,
 		},
+		{
+			in:  `{foo="bar"} |= "foo" |~ "bar" | json | latency >= 10s or foo<5 and bar="t" | line_format "b{{.blip}}"`,
+			out: `downstream<{foo="bar"} |="foo" |~"bar" | json | latency>=10s or foo<5,bar="t"| line_format "b{{.blip}}",shard=0_of_2>++downstream<{foo="bar"} |="foo" |~"bar" | json | latency>=10s or foo<5, bar="t" | line_format "b{{.blip}}",shard=1_of_2>`,
+		},
 		{
 			in:  `sum(rate({foo="bar"}[1m]))`,
 			out: `sum(downstream<sum(rate({foo="bar"}[1m])), shard=0_of_2> ++ downstream<sum(rate({foo="bar"}[1m])), shard=1_of_2>)`,
@@ -162,7 +167,7 @@ func TestMappingStrings(t *testing.T) {
 			mapped, err := m.Map(ast, nilMetrics.shardRecorder())
 			require.Nil(t, err)
 
-			require.Equal(t, tc.out, mapped.String())
+			require.Equal(t, strings.ReplaceAll(tc.out, " ", ""), strings.ReplaceAll(mapped.String(), " ", ""))
 
 		})
 	}
@@ -207,45 +212,38 @@ func TestMapping(t *testing.T) {
 				},
 			},
 		},
-		// todo(cyriltovena) fix
-		// {
-		// 	in: `{foo="bar"} |= "error"`,
-		// 	expr: &ConcatLogSelectorExpr{
-		// 		DownstreamLogSelectorExpr: DownstreamLogSelectorExpr{
-		// 			shard: &astmapper.ShardAnnotation{
-		// 				Shard: 0,
-		// 				Of:    2,
-		// 			},
-		// 			LogSelectorExpr: &filterExpr{
-		// 				match: "error",
-		// 				ty:    labels.MatchEqual,
-		// 				left: &matchersExpr{
-		// 					matchers: []*labels.Matcher{
-		// 						mustNewMatcher(labels.MatchEqual, "foo", "bar"),
-		// 					},
-		// 				},
-		// 			},
-		// 		},
-		// 		next: &ConcatLogSelectorExpr{
-		// 			DownstreamLogSelectorExpr: DownstreamLogSelectorExpr{
-		// 				shard: &astmapper.ShardAnnotation{
-		// 					Shard: 1,
-		// 					Of:    2,
-		// 				},
-		// 				LogSelectorExpr: &filterExpr{
-		// 					match: "error",
-		// 					ty:    labels.MatchEqual,
-		// 					left: &matchersExpr{
-		// 						matchers: []*labels.Matcher{
-		// 							mustNewMatcher(labels.MatchEqual, "foo", "bar"),
-		// 						},
-		// 					},
-		// 				},
-		// 			},
-		// 			next: nil,
-		// 		},
-		// 	},
-		// },
+		{
+			in: `{foo="bar"} |= "error"`,
+			expr: &ConcatLogSelectorExpr{
+				DownstreamLogSelectorExpr: DownstreamLogSelectorExpr{
+					shard: &astmapper.ShardAnnotation{
+						Shard: 0,
+						Of:    2,
+					},
+					LogSelectorExpr: newPipelineExpr(
+						newMatcherExpr([]*labels.Matcher{mustNewMatcher(labels.MatchEqual, "foo", "bar")}),
+						MultiPipelineExpr{
+							newLineFilterExpr(nil, labels.MatchEqual, "error"),
+						},
+					),
+				},
+				next: &ConcatLogSelectorExpr{
+					DownstreamLogSelectorExpr: DownstreamLogSelectorExpr{
+						shard: &astmapper.ShardAnnotation{
+							Shard: 1,
+							Of:    2,
+						},
+						LogSelectorExpr: newPipelineExpr(
+							newMatcherExpr([]*labels.Matcher{mustNewMatcher(labels.MatchEqual, "foo", "bar")}),
+							MultiPipelineExpr{
+								newLineFilterExpr(nil, labels.MatchEqual, "error"),
+							},
+						),
+					},
+					next: nil,
+				},
+			},
+		},
 		{
 			in: `rate({foo="bar"}[5m])`,
 			expr: &ConcatSampleExpr{

pkg/logql/test_utils.go

@@ -90,13 +90,58 @@ outer:
 }
 
 func processStream(in []logproto.Stream, pipeline Pipeline) []logproto.Stream {
-	// todo(cyriltovena)
-	return in
+	resByStream := map[string]*logproto.Stream{}
+
+	for _, stream := range in {
+		for _, e := range stream.Entries {
+			if l, out, ok := pipeline.Process([]byte(e.Line), mustParseLabels(stream.Labels)); ok {
+				var s *logproto.Stream
+				var found bool
+				s, found = resByStream[out.String()]
+				if !found {
+					s = &logproto.Stream{Labels: out.String()}
+					resByStream[out.String()] = s
+				}
+				s.Entries = append(s.Entries, logproto.Entry{
+					Timestamp: e.Timestamp,
+					Line:      string(l),
+				})
+			}
+		}
+	}
+	streams := []logproto.Stream{}
+	for _, stream := range resByStream {
+		streams = append(streams, *stream)
+	}
+	return streams
 }
 
 func processSeries(in []logproto.Stream, pipeline Pipeline, ex SampleExtractor) []logproto.Series {
-	// todo(cyriltovena)
-	return nil
+	resBySeries := map[string]*logproto.Series{}
+
+	for _, stream := range in {
+		for _, e := range stream.Entries {
+			if l, out, ok := pipeline.Process([]byte(e.Line), mustParseLabels(stream.Labels)); ok {
+				f, lbs := ex.Extract(l, out)
+				var s *logproto.Series
+				var found bool
+				s, found = resBySeries[lbs.String()]
+				if !found {
+					s = &logproto.Series{Labels: lbs.String()}
+					resBySeries[lbs.String()] = s
+				}
+				s.Samples = append(s.Samples, logproto.Sample{
+					Timestamp: e.Timestamp.UnixNano(),
+					Value:     f,
+				})
+			}
+		}
+	}
+	series := []logproto.Series{}
+	for _, s := range resBySeries {
+		series = append(series, *s)
+	}
+	return series
 }
 
 func (q MockQuerier) SelectSamples(ctx context.Context, req SelectSampleParams) (iter.SampleIterator, error) {