fix(logql): Make StringLabelfilter work as expected

Consider Log lines

```
msg=hello subqueries=5
msg=hello subqueries=0
msg=hello
```

If you run following logql
```
{foo="bar"}|logfmt|subqueries!=0
```
it returns
```
msg=hello subqueries=5
```

But if you run following logql
```
{foo="bar"}|logfmt|subqueries!="0" # NOTE: "0" as string.
```
it returns
```
msg=hello subqueries=5
msg=hello
```

NOTE: it also returns log lines that doesn't contain `subqueries` label in the first
place.

This cause subtle error if used in metric queries using that `label` in unwrap.

e.g: following logql on above log lines returns
```
sum_over_time({foo="bar"}|logfmt|subqueries!="0"|unwrap subqueries[1m])
```

Returns `pipeline error: 'SampleExtractionErr'`. Because, the lines without `subqueries`
labels are not skipped during `ProcessLine` and during metric extraction, our extractor
sets this `SampleExtractionerror`.

Fixes: #6713

pkg/logql/log/label_filter.go

@@ -342,7 +342,10 @@ func (s *StringLabelFilter) Process(_ int64, line []byte, lbs *LabelsBuilder) ([
 	if s.Name == logqlmodel.ErrorLabel {
 		return line, s.Matches(lbs.GetErr())
 	}
-	v, _ := lbs.Get(s.Name)
+	v, ok := lbs.Get(s.Name)
+	if !ok {
+		return line, false
+	}
 	return line, s.Matches(v)
 }
 

pkg/logql/log/label_filter_test.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/prometheus/prometheus/model/labels"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/grafana/loki/pkg/logqlmodel"
@@ -350,3 +351,32 @@ func TestReduceAndLabelFilter(t *testing.T) {
 		})
 	}
 }
+
+func TestStringLabelFilter(t *testing.T) {
+	tests := []struct {
+		name        string
+		filter      *StringLabelFilter
+		labels      labels.Labels
+		shouldMatch bool
+	}{
+		{
+			name:        "without-label",
+			filter:      NewStringLabelFilter(labels.MustNewMatcher(labels.MatchNotEqual, "subqueries", "0")),
+			labels:      labels.Labels{{"msg", "hello"}}, // no label `subqueries`
+			shouldMatch: false,
+		},
+		{
+			name:        "with-label",
+			filter:      NewStringLabelFilter(labels.MustNewMatcher(labels.MatchNotEqual, "subqueries", "0")),
+			labels:      labels.Labels{{"msg", "hello"}, {"subqueries", "2"}}, // label `subqueries` exist
+			shouldMatch: true,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			_, ok := tc.filter.Process(0, []byte("sample log line"), NewBaseLabelsBuilder().ForLabels(tc.labels, tc.labels.Hash()))
+			assert.Equal(t, tc.shouldMatch, ok)
+		})
+	}
+}

pkg/logql/log/metrics_extraction_test.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/prometheus/prometheus/model/labels"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
@@ -152,6 +153,69 @@ func Test_Extract_ExpectedLabels(t *testing.T) {
 	require.True(t, ok)
 	require.Equal(t, (20 * time.Millisecond).Seconds(), f)
 	require.Equal(t, labels.Labels{{Name: "foo", Value: "json"}}, lbs.Labels())
+
+}
+func TestLabelExtractorWithStages(t *testing.T) {
+
+	// A helper type to check if particular logline should be skipped
+	// during `ProcessLine` or got correct sample value extracted.
+	type checkLine struct {
+		logLine string
+		skip    bool
+		sample  float64
+	}
+
+	tests := []struct {
+		name       string
+		extractor  SampleExtractor
+		checkLines []checkLine
+		shouldFail bool
+	}{
+		{
+			name: "with just logfmt and stringlabelfilter",
+			// {foo="bar"} | logfmt | subqueries != "0" (note: "0", a stringlabelfilter)
+			extractor: mustSampleExtractor(
+				LabelExtractorWithStages("subqueries", ConvertFloat, []string{"foo"}, false, false, []Stage{NewLogfmtParser(), NewStringLabelFilter(labels.MustNewMatcher(labels.MatchNotEqual, "subqueries", "0"))}, NoopStage),
+			),
+			checkLines: []checkLine{
+				{logLine: "msg=hello subqueries=5", skip: false, sample: 5},
+				{logLine: "msg=hello subqueries=0", skip: true},
+				{logLine: "msg=hello ", skip: true}, // log lines doesn't contain the `subqueries` label
+			},
+		},
+		{
+			name: "with just logfmt and numeric labelfilter",
+			// {foo="bar"} | logfmt | subqueries != 0 (note: "0", a numericLabelFilter)
+			extractor: mustSampleExtractor(
+				LabelExtractorWithStages("subqueries", ConvertFloat, []string{"foo"}, false, false, []Stage{NewLogfmtParser(), NewNumericLabelFilter(LabelFilterNotEqual, "subqueries", 0)}, NoopStage),
+			),
+			checkLines: []checkLine{
+				{logLine: "msg=hello subqueries=5", skip: false, sample: 5},
+				{logLine: "msg=hello subqueries=0", skip: true},
+				{logLine: "msg=hello ", skip: true}, // log lines doesn't contain the `subqueries` label
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			for _, line := range tc.checkLines {
+				v, lbs, ok := tc.extractor.ForStream(labels.Labels{{Name: "bar", Value: "foo"}}).ProcessString(0, line.logLine)
+				skipped := !ok
+				assert.Equal(t, line.skip, skipped, "line", line.logLine)
+				if !skipped {
+					assert.Equal(t, line.sample, v)
+
+					// lbs shouldn't have __error__ = SampleExtractionError
+					assert.Empty(t, lbs.Labels())
+					return
+				}
+
+				// if line is skipped, `lbs` will be nil.
+				assert.Nil(t, lbs, "line", line.logLine)
+			}
+		})
+	}
 }
 
 func mustSampleExtractor(ex SampleExtractor, err error) SampleExtractor {