Feature/querysharding ii

pkg/chunkenc/memchunk_test.go

@@ -10,6 +10,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cortexproject/cortex/pkg/util"
 	"github.com/stretchr/testify/assert"
 
 	"github.com/dustin/go-humanize"
@@ -436,7 +437,7 @@ func TestChunkStats(t *testing.T) {
 		t.Fatal(err)
 	}
 	// test on a chunk filling up
-	s := stats.Snapshot(ctx, time.Since(first))
+	s := stats.Snapshot(ctx, time.Since(first), util.Logger)
 	require.Equal(t, int64(expectedSize), s.Summary.TotalBytesProcessed)
 	require.Equal(t, int64(inserted), s.Summary.TotalLinesProcessed)
 
@@ -464,7 +465,7 @@ func TestChunkStats(t *testing.T) {
 	if err := it.Close(); err != nil {
 		t.Fatal(err)
 	}
-	s = stats.Snapshot(ctx, time.Since(first))
+	s = stats.Snapshot(ctx, time.Since(first), util.Logger)
 	require.Equal(t, int64(expectedSize), s.Summary.TotalBytesProcessed)
 	require.Equal(t, int64(inserted), s.Summary.TotalLinesProcessed)
 

pkg/logcli/query/query.go

@@ -119,10 +119,29 @@ func (q *Query) DoLocalQuery(out output.LogOutput, statistics bool, orgID string
 
 	eng := logql.NewEngine(conf.Querier.Engine, querier)
 	var query logql.Query
+
 	if q.isInstant() {
-		query = eng.NewInstantQuery(q.QueryString, q.Start, q.resultsDirection(), uint32(q.Limit))
+		query = eng.Query(logql.NewLiteralParams(
+			q.QueryString,
+			q.Start,
+			q.Start,
+			0,
+			0,
+			q.resultsDirection(),
+			uint32(q.Limit),
+			nil,
+		))
 	} else {
-		query = eng.NewRangeQuery(q.QueryString, q.Start, q.End, q.Step, q.Interval, q.resultsDirection(), uint32(q.Limit))
+		query = eng.Query(logql.NewLiteralParams(
+			q.QueryString,
+			q.Start,
+			q.End,
+			q.Step,
+			q.Interval,
+			q.resultsDirection(),
+			uint32(q.Limit),
+			nil,
+		))
 	}
 
 	// execute the query

pkg/loghttp/params.go

@@ -44,6 +44,10 @@ func direction(r *http.Request) (logproto.Direction, error) {
 	return parseDirection(r.Form.Get("direction"), logproto.BACKWARD)
 }
 
+func shards(r *http.Request) []string {
+	return r.Form["shards"]
+}
+
 func bounds(r *http.Request) (time.Time, time.Time, error) {
 	now := time.Now()
 	start, err := parseTimestamp(r.Form.Get("start"), now.Add(-defaultSince))

pkg/loghttp/query.go

@@ -244,6 +244,7 @@ type RangeQuery struct {
 	Query     string
 	Direction logproto.Direction
 	Limit     uint32
+	Shards    []string
 }
 
 // ParseRangeQuery parses a RangeQuery request from an http request.
@@ -280,6 +281,8 @@ func ParseRangeQuery(r *http.Request) (*RangeQuery, error) {
 		return nil, errNegativeStep
 	}
 
+	result.Shards = shards(r)
+
 	// For safety, limit the number of returned points per timeseries.
 	// This is sufficient for 60s resolution for a week or 1h resolution for a year.
 	if (result.End.Sub(result.Start) / result.Step) > 11000 {

pkg/logproto/extensions.go

@@ -2,6 +2,7 @@ package logproto
 
 import "github.com/prometheus/prometheus/pkg/labels"
 
+// Note, this is not very efficient and use should be minimized as it requires label construction on each comparison
 type SeriesIdentifiers []SeriesIdentifier
 
 func (ids SeriesIdentifiers) Len() int      { return len(ids) }
@@ -10,3 +11,9 @@ func (ids SeriesIdentifiers) Less(i, j int) bool {
 	a, b := labels.FromMap(ids[i].Labels), labels.FromMap(ids[j].Labels)
 	return labels.Compare(a, b) <= 0
 }
+
+type Streams []Stream
+
+func (xs Streams) Len() int           { return len(xs) }
+func (xs Streams) Swap(i, j int)      { xs[i], xs[j] = xs[j], xs[i] }
+func (xs Streams) Less(i, j int) bool { return xs[i].Labels <= xs[j].Labels }

pkg/logproto/logproto.proto

@@ -37,6 +37,8 @@ message QueryRequest {
   google.protobuf.Timestamp end = 4 [(gogoproto.stdtime) = true, (gogoproto.nullable) = false];
   Direction direction = 5;
   reserved 6;
+  repeated string shards = 7 [(gogoproto.jsontag) = "shards,omitempty"];
+
 }
 
 enum Direction {