Fixes a bug in MatrixStepper when sharding queries.

Since we split correctly metric queries, this bug has show itself.
Basically we were not correctly stepping through time. We should always start from start, add the step until the start is after the end.

For more read: https://www.robustperception.io/step-and-query_range

Fixes #3541

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

pkg/logcli/query/query.go

@@ -58,7 +58,6 @@ type Query struct {
 
 // DoQuery executes the query and prints out the results
 func (q *Query) DoQuery(c client.Client, out output.LogOutput, statistics bool) {
-
 	if q.LocalConfig != "" {
 		if err := q.DoLocalQuery(out, statistics, c.GetOrgID()); err != nil {
 			log.Fatalf("Query failed: %+v", err)
@@ -149,7 +148,6 @@ func (q *Query) DoQuery(c client.Client, out output.LogOutput, statistics bool)
 
 		}
 	}
-
 }
 
 func (q *Query) printResult(value loghttp.ResultValue, out output.LogOutput, lastEntry []*loghttp.Entry) (int, []*loghttp.Entry) {
@@ -172,7 +170,6 @@ func (q *Query) printResult(value loghttp.ResultValue, out output.LogOutput, las
 
 // DoLocalQuery executes the query against the local store using a Loki configuration file.
 func (q *Query) DoLocalQuery(out output.LogOutput, statistics bool, orgID string) error {
-
 	var conf loki.Config
 	conf.RegisterFlags(flag.CommandLine)
 	if q.LocalConfig == "" {
@@ -255,7 +252,7 @@ func (q *Query) SetInstant(time time.Time) {
 }
 
 func (q *Query) isInstant() bool {
-	return q.Start == q.End
+	return q.Start == q.End && q.Step == 0
 }
 
 func (q *Query) printStream(streams loghttp.Streams, out output.LogOutput, lastEntry []*loghttp.Entry) (int, []*loghttp.Entry) {
@@ -369,7 +366,6 @@ func (q *Query) printMatrix(matrix loghttp.Matrix) {
 	// it gives us more flexibility with regard to output types in the future.  initially we are supporting just formatted json but eventually
 	// we might add output options such as render to an image file on disk
 	bytes, err := json.MarshalIndent(matrix, "", "  ")
-
 	if err != nil {
 		log.Fatalf("Error marshalling matrix: %v", err)
 	}
@@ -379,7 +375,6 @@ func (q *Query) printMatrix(matrix loghttp.Matrix) {
 
 func (q *Query) printVector(vector loghttp.Vector) {
 	bytes, err := json.MarshalIndent(vector, "", "  ")
-
 	if err != nil {
 		log.Fatalf("Error marshalling vector: %v", err)
 	}
@@ -389,7 +384,6 @@ func (q *Query) printVector(vector loghttp.Vector) {
 
 func (q *Query) printScalar(scalar loghttp.Scalar) {
 	bytes, err := json.MarshalIndent(scalar, "", "  ")
-
 	if err != nil {
 		log.Fatalf("Error marshalling scalar: %v", err)
 	}

pkg/logql/matrix.go

@@ -33,7 +33,7 @@ func NewMatrixStepper(start, end time.Time, step time.Duration, m promql.Matrix)
 
 func (m *MatrixStepper) Next() (bool, int64, promql.Vector) {
 	m.ts = m.ts.Add(m.step)
-	if !m.ts.Before(m.end) {
+	if m.ts.After(m.end) {
 		return false, 0, nil
 	}
 

pkg/logql/matrix_test.go

@@ -20,7 +20,7 @@ func TestMatrixStepper(t *testing.T) {
 		promql.Series{
 			Metric: labels.Labels{{Name: "foo", Value: "bar"}},
 			Points: []promql.Point{
-				{T: start.UnixNano() / int64(step), V: 0},
+				{T: start.UnixNano(), V: 0},
 				{T: start.Add(step).UnixNano() / int64(time.Millisecond), V: 1},
 				{T: start.Add(2*step).UnixNano() / int64(time.Millisecond), V: 2},
 				{T: start.Add(3*step).UnixNano() / int64(time.Millisecond), V: 3},
@@ -42,11 +42,11 @@ func TestMatrixStepper(t *testing.T) {
 	expected := []promql.Vector{
 		{
 			promql.Sample{
-				Point:  promql.Point{T: start.UnixNano() / int64(step), V: 0},
+				Point:  promql.Point{T: start.UnixNano(), V: 0},
 				Metric: labels.Labels{{Name: "foo", Value: "bar"}},
 			},
 			promql.Sample{
-				Point:  promql.Point{T: start.UnixNano() / int64(step), V: 0},
+				Point:  promql.Point{T: start.UnixNano(), V: 0},
 				Metric: labels.Labels{{Name: "bazz", Value: "buzz"}},
 			},
 		},
@@ -100,9 +100,19 @@ func TestMatrixStepper(t *testing.T) {
 				Metric: labels.Labels{{Name: "bazz", Value: "buzz"}},
 			},
 		},
+		{
+			promql.Sample{
+				Point:  promql.Point{T: start.Add(6*step).UnixNano() / int64(time.Millisecond), V: 0},
+				Metric: labels.Labels{{Name: "foo", Value: "bar"}},
+			},
+			promql.Sample{
+				Point:  promql.Point{T: start.Add(6*step).UnixNano() / int64(time.Millisecond), V: 0},
+				Metric: labels.Labels{{Name: "bazz", Value: "buzz"}},
+			},
+		},
 	}
 
-	for i := 0; i < int(end.Sub(start)/step); i++ {
+	for i := 0; i <= int(end.Sub(start)/step); i++ {
 		ok, ts, vec := s.Next()
 		require.Equal(t, ok, true)
 		require.Equal(t, start.Add(step*time.Duration(i)).UnixNano()/int64(time.Millisecond), ts)
@@ -113,3 +123,33 @@ func TestMatrixStepper(t *testing.T) {
 
 	require.Equal(t, ok, false)
 }
+
+func Test_SingleStepMatrix(t *testing.T) {
+	var (
+		start = time.Unix(0, 0)
+		end   = time.Unix(0, 0)
+		step  = time.Second
+	)
+
+	m := promql.Matrix{
+		promql.Series{
+			Metric: labels.Labels{},
+			Points: []promql.Point{
+				{T: start.UnixNano(), V: 10},
+			},
+		},
+	}
+
+	s := NewMatrixStepper(start, end, step, m)
+
+	ok, ts, vec := s.Next()
+	require.True(t, ok)
+	require.Equal(t, start.UnixNano(), ts)
+	require.Equal(t, promql.Vector{promql.Sample{
+		Point:  promql.Point{T: start.UnixNano(), V: 10},
+		Metric: labels.Labels{},
+	}}, vec)
+
+	ok, _, _ = s.Next()
+	require.False(t, ok)
+}