feat: reject filter queries to /patterns endpoint

pkg/logql/range_vector.go

@@ -108,20 +108,20 @@ func NewBatchRangeVectorIterator(
 		end:      end,
 		current:  start - step, // first loop iteration will set it to start
 		offset:   offset,
-    metrics:  map[string]labels.Labels{},
-    window:   map[string]*promql.Series{},
+		metrics:  map[string]labels.Labels{},
+		window:   map[string]*promql.Series{},
 		agg:      agg,
 	}
 }
 
 func (r *batchRangeVectorIterator) Next() bool {
 	// slides the range window to the next position
-	r.current = r.current + r.step // first current will be 5 min before start
+	r.current = r.current + r.step
 	if r.current > r.end {
 		return false
 	}
 	rangeEnd := r.current
-	rangeStart := rangeEnd - r.selRange // in nanoseconds
+	rangeStart := rangeEnd - r.selRange
 	// load samples
 	r.popBack(rangeStart)
 	r.load(rangeStart, rangeEnd)

pkg/loki/loki.go

@@ -331,7 +331,7 @@ type Loki struct {
 	distributor               *distributor.Distributor
 	Ingester                  ingester.Interface
 	PatternIngester           *pattern.Ingester
-	PatternRingClient         *pattern.RingClient
+	PatternRingClient         pattern.RingClient
 	Querier                   querier.Querier
 	cacheGenerationLoader     queryrangebase.CacheGenNumberLoader
 	querierAPI                *querier.QuerierAPI

pkg/pattern/ingester_querier.go

@@ -4,10 +4,8 @@ import (
 	"context"
 	"errors"
 	"math"
-	"net/http"
 
 	"github.com/go-kit/log"
-	"github.com/grafana/dskit/httpgrpc"
 	"github.com/grafana/dskit/ring"
 	"github.com/prometheus/client_golang/prometheus"
 
@@ -26,14 +24,14 @@ type IngesterQuerier struct {
 	cfg    Config
 	logger log.Logger
 
-	ringClient *RingClient
+	ringClient RingClient
 
 	registerer prometheus.Registerer
 }
 
 func NewIngesterQuerier(
 	cfg Config,
-	ringClient *RingClient,
+	ringClient RingClient,
 	metricsNamespace string,
 	registerer prometheus.Registerer,
 	logger log.Logger,
@@ -48,21 +46,31 @@ func NewIngesterQuerier(
 
 func (q *IngesterQuerier) Patterns(ctx context.Context, req *logproto.QueryPatternsRequest) (*logproto.QueryPatternsResponse, error) {
 	// validate that a supported query was provided
-	// TODO(twhitney): validate metric queries don't have filters
 	var expr syntax.Expr
 	_, err := syntax.ParseMatchers(req.Query, true)
 	if err != nil {
 		expr, err = syntax.ParseSampleExpr(req.Query)
 		if err != nil {
-			return nil, httpgrpc.Errorf(http.StatusBadRequest, ErrParseQuery.Error())
+			return nil, ErrParseQuery
 		}
 
+		var selector syntax.LogSelectorExpr
 		switch expr.(type) {
-		case *syntax.VectorAggregationExpr, *syntax.RangeAggregationExpr:
-			break
+		case *syntax.VectorAggregationExpr:
+			selector, err = expr.(*syntax.VectorAggregationExpr).Selector()
+		case *syntax.RangeAggregationExpr:
+			selector, err = expr.(*syntax.RangeAggregationExpr).Selector()
 		default:
 			return nil, ErrParseQuery
 		}
+
+		if err != nil {
+			return nil, err
+		}
+
+		if selector == nil || selector.HasFilter() {
+			return nil, ErrParseQuery
+		}
 	}
 
 	resps, err := q.forAllIngesters(ctx, func(_ context.Context, client logproto.PatternClient) (interface{}, error) {
@@ -118,7 +126,7 @@ func prunePatterns(
 
 // ForAllIngesters runs f, in parallel, for all ingesters
 func (q *IngesterQuerier) forAllIngesters(ctx context.Context, f func(context.Context, logproto.PatternClient) (interface{}, error)) ([]ResponseFromIngesters, error) {
-	replicationSet, err := q.ringClient.ring.GetReplicationSetForOperation(ring.Read)
+	replicationSet, err := q.ringClient.Ring().GetReplicationSetForOperation(ring.Read)
 	if err != nil {
 		return nil, err
 	}
@@ -137,7 +145,7 @@ func (q *IngesterQuerier) forGivenIngesters(ctx context.Context, replicationSet
 		// Nothing here
 	}
 	results, err := ring.DoUntilQuorum(ctx, replicationSet, cfg, func(ctx context.Context, ingester *ring.InstanceDesc) (ResponseFromIngesters, error) {
-		client, err := q.ringClient.pool.GetClientFor(ingester.Addr)
+		client, err := q.ringClient.Pool().GetClientFor(ingester.Addr)
 		if err != nil {
 			return ResponseFromIngesters{addr: ingester.Addr}, err
 		}

pkg/pattern/ingester_querier_test.go

@@ -2,11 +2,17 @@ package pattern
 
 import (
 	"bufio"
+	"context"
 	"os"
 	"testing"
+	"time"
 
+	"github.com/go-kit/log"
 	"github.com/stretchr/testify/require"
 
+	"github.com/grafana/dskit/ring"
+	ring_client "github.com/grafana/dskit/ring/client"
+	"github.com/grafana/dskit/services"
 	"github.com/grafana/loki/v3/pkg/logproto"
 )
 
@@ -39,3 +45,155 @@ func Test_prunePatterns(t *testing.T) {
 
 	require.Equal(t, expectedPatterns, patterns)
 }
+
+func Test_Patterns(t *testing.T) {
+	t.Run("it rejects metric queries with filters", func(t *testing.T) {
+		q := &IngesterQuerier{
+			cfg:        Config{},
+			logger:     log.NewNopLogger(),
+			ringClient: &fakeRingClient{},
+			registerer: nil,
+		}
+		for _, query := range []string{
+			`count_over_time({foo="bar"} |= "baz" [5m])`,
+			`count_over_time({foo="bar"} != "baz" [5m])`,
+			`count_over_time({foo="bar"} =~ "baz" [5m])`,
+			`count_over_time({foo="bar"} !~ "baz" [5m])`,
+			`count_over_time({foo="bar"} | logfmt | color=blue [5m])`,
+			`sum(count_over_time({foo="bar"} |= "baz" [5m]))`,
+			`sum by label(count_over_time({foo="bar"} |= "baz" [5m]))`,
+			`bytes_over_time({foo="bar"} |= "baz" [5m])`,
+		} {
+			_, err := q.Patterns(
+				context.Background(),
+				&logproto.QueryPatternsRequest{
+					Query: query,
+				},
+			)
+			require.Error(t, err, query)
+			require.ErrorIs(t, err, ErrParseQuery)
+
+		}
+	})
+
+	t.Run("accepts log selector queries and count and bytes metric queries", func(t *testing.T) {
+		q := &IngesterQuerier{
+			cfg:        Config{},
+			logger:     log.NewNopLogger(),
+			ringClient: &fakeRingClient{},
+			registerer: nil,
+		}
+		for _, query := range []string{
+			`{foo="bar"}`,
+			`count_over_time({foo="bar"}[5m])`,
+			`bytes_over_time({foo="bar"}[5m])`,
+			`sum(count_over_time({foo="bar"}[5m]))`,
+			`sum(bytes_over_time({foo="bar"}[5m]))`,
+			`sum by (level)(count_over_time({foo="bar"}[5m]))`,
+			`sum by (level)(bytes_over_time({foo="bar"}[5m]))`,
+		} {
+			_, err := q.Patterns(
+				context.Background(),
+				&logproto.QueryPatternsRequest{
+					Query: query,
+				},
+			)
+			require.NoError(t, err, query)
+		}
+	})
+}
+
+type fakeRingClient struct{}
+
+func (f *fakeRingClient) Pool() *ring_client.Pool {
+	panic("not implemented")
+}
+
+func (f *fakeRingClient) StartAsync(ctx context.Context) error {
+	panic("not implemented")
+}
+
+func (f *fakeRingClient) AwaitRunning(ctx context.Context) error {
+	panic("not implemented")
+}
+
+func (f *fakeRingClient) StopAsync() {
+	panic("not implemented")
+}
+
+func (f *fakeRingClient) AwaitTerminated(ctx context.Context) error {
+	panic("not implemented")
+}
+
+func (f *fakeRingClient) FailureCase() error {
+	panic("not implemented")
+}
+
+func (f *fakeRingClient) State() services.State {
+	panic("not implemented")
+}
+
+func (f *fakeRingClient) AddListener(listener services.Listener) {
+	panic("not implemented")
+}
+
+func (f *fakeRingClient) Ring() ring.ReadRing {
+	return &fakeRing{}
+}
+
+type fakeRing struct{}
+
+func (f *fakeRing) Get(
+	key uint32,
+	op ring.Operation,
+	bufDescs []ring.InstanceDesc,
+	bufHosts []string,
+	bufZones []string,
+) (ring.ReplicationSet, error) {
+	panic("not implemented")
+}
+
+func (f *fakeRing) GetAllHealthy(op ring.Operation) (ring.ReplicationSet, error) {
+	panic("not implemented")
+}
+
+func (f *fakeRing) GetReplicationSetForOperation(op ring.Operation) (ring.ReplicationSet, error) {
+	return ring.ReplicationSet{}, nil
+}
+
+func (f *fakeRing) ReplicationFactor() int {
+	panic("not implemented")
+}
+
+func (f *fakeRing) InstancesCount() int {
+	panic("not implemented")
+}
+
+func (f *fakeRing) ShuffleShard(identifier string, size int) ring.ReadRing {
+	panic("not implemented")
+}
+
+func (f *fakeRing) GetInstanceState(instanceID string) (ring.InstanceState, error) {
+	panic("not implemented")
+}
+
+func (f *fakeRing) ShuffleShardWithLookback(
+	identifier string,
+	size int,
+	lookbackPeriod time.Duration,
+	now time.Time,
+) ring.ReadRing {
+	panic("not implemented")
+}
+
+func (f *fakeRing) HasInstance(instanceID string) bool {
+	panic("not implemented")
+}
+
+func (f *fakeRing) CleanupShuffleShardCache(identifier string) {
+	panic("not implemented")
+}
+
+func (f *fakeRing) GetTokenRangesForInstance(instanceID string) (ring.TokenRanges, error) {
+	panic("not implemented")
+}

pkg/pattern/ingester_test.go

@@ -93,7 +93,7 @@ func TestInstancePushQuery(t *testing.T) {
 
 	t.Run("test count_over_time samples", func(t *testing.T) {
 		lbs := labels.New(labels.Label{Name: "test", Value: "test"})
-		inst, err := newInstance("foo", log.NewNopLogger())
+		inst, err := newInstance("foo", log.NewNopLogger(), nil)
 		require.NoError(t, err)
 
 		err = inst.Push(context.Background(), &push.PushRequest{
@@ -185,7 +185,7 @@ func TestInstancePushQuery(t *testing.T) {
 
 	t.Run("test bytes_over_time samples", func(t *testing.T) {
 		lbs := labels.New(labels.Label{Name: "test", Value: "test"})
-		inst, err := newInstance("foo", log.NewNopLogger())
+		inst, err := newInstance("foo", log.NewNopLogger(), nil)
 		require.NoError(t, err)
 
 		err = inst.Push(context.Background(), &push.PushRequest{

pkg/pattern/instance_test.go

@@ -28,7 +28,7 @@ func TestInstance_QuerySample(t *testing.T) {
 		Step:  oneMin,
 	}
 
-	instance, err := newInstance("test", log.NewNopLogger())
+	instance, err := newInstance("test", log.NewNopLogger(), nil)
 	require.NoError(t, err)
 
 	labels := model.LabelSet{

pkg/pattern/metric/evaluator.go

@@ -11,6 +11,7 @@ import (
 	"github.com/grafana/loki/v3/pkg/logql"
 	"github.com/grafana/loki/v3/pkg/logql/syntax"
 	"github.com/grafana/loki/v3/pkg/pattern/iter"
+	"github.com/grafana/loki/v3/pkg/storage/chunk/cache/resultscache"
 	"github.com/pkg/errors"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/model/labels"
@@ -48,9 +49,9 @@ func ExtractMetricType(expr syntax.SampleExpr) (MetricType, 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 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,
@@ -161,7 +162,7 @@ func (ev *DefaultEvaluatorFactory) NewStepEvaluator(
 	}
 }
 
-// Need to create our own StepEvaluator since we only support bytes and count over time, 
+// Need to create our own StepEvaluator since we only support bytes and count over time,
 // and always sum to get those values. In order to accomplish this we need control over the
 // aggregation operation..
 func NewPatternSampleRangeAggEvaluator(
@@ -199,7 +200,7 @@ func newRangeVectorIterator(
 	// TODO(twhitney): do I need a streaming aggregator?
 	// if so the aggregator is going to make this
 	// a bit of a bad time, as there's currently no
-  // way to provide a custom one.
+	// way to provide a custom one.
 	//
 	// var overlap bool
 	// if selRange >= step && start != end {
@@ -222,7 +223,7 @@ func newRangeVectorIterator(
 	// 	}, nil
 	// }
 
-  // always sum
+	// always sum
 	aggregator := logql.BatchRangeVectorAggregator(func(samples []promql.FPoint) float64 {
 		sum := 0.0
 		for _, v := range samples {
@@ -263,7 +264,6 @@ func (s *SeriesToSampleIterator) Next() bool {
 
 	current, rest := s.floats[0], s.floats[1:]
 
-  //Is timestamp the correct unit here?
 	s.curTs = current.T
 	s.cur = current.F
 
@@ -352,3 +352,7 @@ func (p *paramCompat) GetExpression() syntax.Expr {
 func (p *paramCompat) GetStoreChunks() *logproto.ChunkRefGroup {
 	return nil
 }
+
+func (p *paramCompat) CachingOptions() (res resultscache.CachingOptions) {
+	return
+}