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

[query] Fix graphite functions, aggregation bug #2549

Merged
merged 20 commits into from
Aug 23, 2020
Merged

[query] Fix graphite functions, aggregation bug #2549

Show file tree
Hide file tree
Changes from 16 commits
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
57c9985
[query] Fix graphite panic in movingMedian, re-enabled some tests
arnikola Aug 21, 2020
ed90f14
Merge branch 'master' into arnikola/moving-median
arnikola Aug 21, 2020
1f16a8f
Merge branch 'master' into arnikola/moving-median
robskillington Aug 21, 2020
4c26204
Import order
arnikola Aug 21, 2020
c340f1b
Merge branch 'arnikola/moving-median' of github.com:m3db/m3 into arni…
arnikola Aug 21, 2020
218b671
Linting
arnikola Aug 21, 2020
fcc9c80
working state
arnikola Aug 22, 2020
83aeabb
stuff
arnikola Aug 22, 2020
33ed9b8
Cleanup
arnikola Aug 22, 2020
31f5999
Cleanup, additional tests
arnikola Aug 22, 2020
ab97310
Cleanup
arnikola Aug 22, 2020
2427b12
Merge branch 'master' into arnikola/moving-median
arnikola Aug 22, 2020
f7e8e2b
Minor cleanup
arnikola Aug 22, 2020
0d658d5
Merge branch 'arnikola/moving-median' of github.com:m3db/m3 into arni…
arnikola Aug 22, 2020
8923dca
Fix test
arnikola Aug 22, 2020
70faf7e
Revert accidentally changed test value
arnikola Aug 22, 2020
efa6a97
PR response
arnikola Aug 22, 2020
117bf9a
Fix to some tests
arnikola Aug 23, 2020
62dfe0d
Test fix
arnikola Aug 23, 2020
1093151
Fix test log
arnikola Aug 23, 2020
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
157 changes: 155 additions & 2 deletions src/aggregator/aggregator/aggregator_mock.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion src/aggregator/generated/mocks/generate.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
// THE SOFTWARE.

// mockgen rules for generating mocks for exported interfaces (reflection mode).
//go:generate sh -c "mockgen -package=aggregator github.com/m3db/m3/src/aggregator/aggregator ElectionManager,FlushTimesManager,PlacementManager | genclean -pkg github.com/m3db/m3/src/aggregator/aggregator -out $GOPATH/src/github.com/m3db/m3/src/aggregator/aggregator/aggregator_mock.go"
//go:generate sh -c "mockgen -package=aggregator github.com/m3db/m3/src/aggregator/aggregator Aggregator,ElectionManager,FlushTimesManager,PlacementManager | genclean -pkg github.com/m3db/m3/src/aggregator/aggregator -out $GOPATH/src/github.com/m3db/m3/src/aggregator/aggregator/aggregator_mock.go"
//go:generate sh -c "mockgen -package=client github.com/m3db/m3/src/aggregator/client Client,AdminClient | genclean -pkg github.com/m3db/m3/src/aggregator/client -out $GOPATH/src/github.com/m3db/m3/src/aggregator/client/client_mock.go"
//go:generate sh -c "mockgen -package=handler github.com/m3db/m3/src/aggregator/aggregator/handler Handler | genclean -pkg github.com/m3db/m3/src/aggregator/aggregator/handler -out $GOPATH/src/github.com/m3db/m3/src/aggregator/aggregator/handler/handler_mock.go"
//go:generate sh -c "mockgen -package=runtime github.com/m3db/m3/src/aggregator/runtime OptionsWatcher | genclean -pkg github.com/m3db/m3/src/aggregator/runtime -out $GOPATH/src/github.com/m3db/m3/src/aggregator/runtime/runtime_mock.go"
Expand Down
107 changes: 106 additions & 1 deletion src/cmd/services/m3coordinator/downsample/flush_handler_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,14 +22,17 @@ package downsample

import (
"bytes"
"sync"
"testing"

"github.com/m3db/m3/src/metrics/metric/aggregated"
"github.com/m3db/m3/src/metrics/metric/id"
"github.com/m3db/m3/src/metrics/policy"
"github.com/m3db/m3/src/query/models"
"github.com/m3db/m3/src/query/storage/mock"
"github.com/m3db/m3/src/x/ident"
"github.com/m3db/m3/src/x/instrument"
"github.com/m3db/m3/src/x/pool"
"github.com/m3db/m3/src/x/serialize"
xsync "github.com/m3db/m3/src/x/sync"
xtest "github.com/m3db/m3/src/x/test"
Expand All @@ -41,7 +44,7 @@ import (
)

func TestDownsamplerFlushHandlerCopiesTags(t *testing.T) {
ctrl := gomock.NewController(t)
ctrl := xtest.NewController(t)
defer ctrl.Finish()

store := mock.NewMockStorage()
Expand Down Expand Up @@ -104,3 +107,105 @@ func TestDownsamplerFlushHandlerCopiesTags(t *testing.T) {
assert.False(t, xtest.ByteSlicesBackedBySameData(tagName, tag.Name))
assert.False(t, xtest.ByteSlicesBackedBySameData(tagValue, tag.Value))
}

func mustEncodedTags(first string, encPool serialize.TagEncoderPool) []byte {
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: Name “mustGraphiteEncodedTags(...)” to be clear it’s forming graphite set of tags?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sg, nothing really special about them being graphite here, was just working off the repro for test data :p

enc := encPool.Get()
defer enc.Finalize()

if err := enc.Encode(ident.MustNewTagStringsIterator(
"__g0__", first,
"__g1__", "y",
"__g2__", "z",
string(MetricsOptionIDSchemeTagName), string(GraphiteIDSchemeTagValue),
)); err != nil {
panic(err.Error())
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Instead of panic, why not remove the “must” prefix and make the test take a “t *testing.T” as first arg and then “require.NoError(t, err)” like we do with some other test methods?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The main motivation here was it already using a call to MustNewTagStringsIterator so it had the panic semantics already, happy to go either way though

}

data, _ := enc.Data()
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: Perhaps take a “t *testing.T” and check that second arg “ok” is true?

data, ok := enc.Data()
require.True(t, ok)

return append(make([]byte, 0, data.Len()), data.Bytes()...)
}

func TestDownsamplerFlushHandlerHighConcurrencyNoTagMixing(t *testing.T) {
ctrl := xtest.NewController(t)
defer ctrl.Finish()

store := mock.NewMockStorage()

size := 10
decodeOpts := serialize.NewTagDecoderOptions(serialize.TagDecoderOptionsConfig{
CheckBytesWrapperPoolSize: &size,
})

poolOpts := pool.NewObjectPoolOptions()
tagDecoderPool := serialize.NewTagDecoderPool(decodeOpts, poolOpts)
tagDecoderPool.Init()

pool := serialize.NewMetricTagsIteratorPool(tagDecoderPool, poolOpts)
pool.Init()

workers := xsync.NewWorkerPool(1)
workers.Init()

instrumentOpts := instrument.NewOptions()

handler := newDownsamplerFlushHandler(store, pool,
workers, models.NewTagOptions(), instrumentOpts)
writer, err := handler.NewWriter(tally.NoopScope)
require.NoError(t, err)

encodeOpts := serialize.NewTagEncoderOptions()
encPool := serialize.NewTagEncoderPool(encodeOpts, poolOpts)
encPool.Init()

xBytes := mustEncodedTags("x", encPool)
fooBytes := mustEncodedTags("foo", encPool)

var wg sync.WaitGroup
for i := 0; i < 100; i++ {
wg.Add(1)
xData := append(make([]byte, 0, len(xBytes)), xBytes...)
fooData := append(make([]byte, 0, len(fooBytes)), fooBytes...)
go func() {
defer wg.Done()
err := writer.Write(aggregated.ChunkedMetricWithStoragePolicy{
ChunkedMetric: aggregated.ChunkedMetric{
ChunkedID: id.ChunkedID{Data: xData},
TimeNanos: 123,
Value: 42.42,
},
StoragePolicy: policy.MustParseStoragePolicy("1s:1d"),
})
require.NoError(t, err)

err = writer.Write(aggregated.ChunkedMetricWithStoragePolicy{
ChunkedMetric: aggregated.ChunkedMetric{
ChunkedID: id.ChunkedID{Data: fooData},
TimeNanos: 123,
Value: 42.42,
},
StoragePolicy: policy.MustParseStoragePolicy("1s:1d"),
})
require.NoError(t, err)
}()
}

wg.Wait()
// Wait for flush
err = writer.Flush()
require.NoError(t, err)

// Inspect the write
writes := store.Writes()
require.Equal(t, 200, len(writes))

seenMap := make(map[string]int, 10)
for _, w := range writes {
str := w.Tags().String()
seenMap[str] = seenMap[str] + 1
}

assert.Equal(t, map[string]int{
"__g0__: foo, __g1__: y, __g2__: z": 100,
"__g0__: x, __g1__: y, __g2__: z": 100,
}, seenMap)
}
7 changes: 4 additions & 3 deletions src/cmd/services/m3coordinator/downsample/metrics_appender.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,10 @@ func newMetricsAppenderPool(opts pool.ObjectPoolOptions) *metricsAppenderPool {
}

func (p *metricsAppenderPool) Get() *metricsAppender {
return p.pool.Get().(*metricsAppender)
appender := p.pool.Get().(*metricsAppender)
// NB: reset appender.
appender.NextMetric()
return appender
}

func (p *metricsAppenderPool) Put(v *metricsAppender) {
Expand Down Expand Up @@ -177,7 +180,6 @@ func (a *metricsAppender) SamplesAppender(opts SampleAppenderOptions) (SamplesAp

a.multiSamplesAppender.reset()
unownedID := data.Bytes()

// Match policies and rollups and build samples appender
id := a.metricTagsIteratorPool.Get()
id.Reset(unownedID)
Expand Down Expand Up @@ -353,7 +355,6 @@ func (a *metricsAppender) SamplesAppender(opts SampleAppenderOptions) (SamplesAp

a.debugLogMatch("downsampler applying matched rollup rule",
debugLogMatchOptions{Meta: rollup.Metadatas, RollupID: rollup.ID})

a.multiSamplesAppender.addSamplesAppender(samplesAppender{
agg: a.agg,
clientRemote: a.clientRemote,
Expand Down
Loading