Skip to content

Commit

Permalink
Fixes a bug in MatrixStepper when sharding queries.
Browse files Browse the repository at this point in the history
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 grafana#3541

Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com>
  • Loading branch information
cyriltovena committed Mar 26, 2021
1 parent 37a7189 commit 30dc2d6
Show file tree
Hide file tree
Showing 3 changed files with 46 additions and 12 deletions.
8 changes: 1 addition & 7 deletions pkg/logcli/query/query.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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) {
Expand All @@ -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 == "" {
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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)
}
Expand All @@ -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)
}
Expand All @@ -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)
}
Expand Down
2 changes: 1 addition & 1 deletion pkg/logql/matrix.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
}

Expand Down
48 changes: 44 additions & 4 deletions pkg/logql/matrix_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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},
Expand All @@ -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"}},
},
},
Expand Down Expand Up @@ -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)
Expand All @@ -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)
}

0 comments on commit 30dc2d6

Please sign in to comment.