Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

test: Fix race condition in LogQL test #12247

Merged
merged 3 commits into from
Mar 19, 2024
Merged

test: Fix race condition in LogQL test #12247

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
092d7f8
Revert "fix: correct use of atomic int (#12225)"
bboreham Mar 18, 2024
9faac32
Revert "fix: data race in seriesIterator, use atomic int (#12223)"
bboreham Mar 18, 2024
77cd95f
test: avoid sharing iterators across queries in pkg/logql
bboreham Mar 18, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 5 additions & 6 deletions pkg/iter/sample_iterator.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@ package iter
import (
"container/heap"
"context"
"go.uber.org/atomic"
"io"
"sync"

Expand Down Expand Up @@ -522,7 +521,7 @@ func NewSampleQueryResponseIterator(resp *logproto.SampleQueryResponse) SampleIt
}

type seriesIterator struct {
i *atomic.Int32
i int
series logproto.Series
}

Expand Down Expand Up @@ -568,14 +567,14 @@ func NewMultiSeriesIterator(series []logproto.Series) SampleIterator {
// NewSeriesIterator iterates over sample in a series.
func NewSeriesIterator(series logproto.Series) SampleIterator {
return &seriesIterator{
i: atomic.NewInt32(-1),
i: -1,
series: series,
}
}

func (i *seriesIterator) Next() bool {
tmp := i.i.Add(1)
return int(tmp) < len(i.series.Samples)
i.i++
return i.i < len(i.series.Samples)
}

func (i *seriesIterator) Error() error {
Expand All @@ -591,7 +590,7 @@ func (i *seriesIterator) StreamHash() uint64 {
}

func (i *seriesIterator) Sample() logproto.Sample {
return i.series.Samples[i.i.Load()]
return i.series.Samples[i.i]
}

func (i *seriesIterator) Close() error {
Expand Down
32 changes: 19 additions & 13 deletions pkg/logql/engine_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2383,16 +2383,16 @@ func TestEngine_LogsInstantQuery_Vector(t *testing.T) {
}

type errorIteratorQuerier struct {
samples []iter.SampleIterator
entries []iter.EntryIterator
samples func() []iter.SampleIterator
entries func() []iter.EntryIterator
}

func (e errorIteratorQuerier) SelectLogs(_ context.Context, p SelectLogParams) (iter.EntryIterator, error) {
return iter.NewSortEntryIterator(e.entries, p.Direction), nil
return iter.NewSortEntryIterator(e.entries(), p.Direction), nil
}

func (e errorIteratorQuerier) SelectSamples(_ context.Context, _ SelectSampleParams) (iter.SampleIterator, error) {
return iter.NewSortSampleIterator(e.samples), nil
return iter.NewSortSampleIterator(e.samples()), nil
}

func TestStepEvaluator_Error(t *testing.T) {
Expand All @@ -2406,9 +2406,11 @@ func TestStepEvaluator_Error(t *testing.T) {
"rangeAggEvaluator",
`count_over_time({app="foo"}[1m])`,
&errorIteratorQuerier{
samples: []iter.SampleIterator{
iter.NewSeriesIterator(newSeries(testSize, identity, `{app="foo"}`)),
NewErrorSampleIterator(),
samples: func() []iter.SampleIterator {
return []iter.SampleIterator{
iter.NewSeriesIterator(newSeries(testSize, identity, `{app="foo"}`)),
NewErrorSampleIterator(),
}
},
},
ErrMock,
Expand All @@ -2417,9 +2419,11 @@ func TestStepEvaluator_Error(t *testing.T) {
"stream",
`{app="foo"}`,
&errorIteratorQuerier{
entries: []iter.EntryIterator{
iter.NewStreamIterator(newStream(testSize, identity, `{app="foo"}`)),
NewErrorEntryIterator(),
entries: func() []iter.EntryIterator {
return []iter.EntryIterator{
iter.NewStreamIterator(newStream(testSize, identity, `{app="foo"}`)),
NewErrorEntryIterator(),
}
},
},
ErrMock,
Expand All @@ -2428,9 +2432,11 @@ func TestStepEvaluator_Error(t *testing.T) {
"binOpStepEvaluator",
`count_over_time({app="foo"}[1m]) / count_over_time({app="foo"}[1m])`,
&errorIteratorQuerier{
samples: []iter.SampleIterator{
iter.NewSeriesIterator(newSeries(testSize, identity, `{app="foo"}`)),
NewErrorSampleIterator(),
samples: func() []iter.SampleIterator {
return []iter.SampleIterator{
iter.NewSeriesIterator(newSeries(testSize, identity, `{app="foo"}`)),
NewErrorSampleIterator(),
}
},
},
ErrMockMultiple,
Expand Down
Loading