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

sidecar: fix issue #396: split response into chunks no bigger than 2^16 samples #718

Merged
merged 3 commits into from
Jan 9, 2019
Merged

sidecar: fix issue #396: split response into chunks no bigger than 2^16 samples #718

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
cc862d1
fixes issue #396, split response into chunks no bigger than 2^16 to a…
trevorwhitney Jan 8, 2019
b2cf675
use single underlying array for chunkSamples
Jan 9, 2019
6b7af4b
make SplitSamplesIntoChunksWithMaxSizeOfUint16_e2e use 3 chunks
Jan 9, 2019
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
43 changes: 33 additions & 10 deletions pkg/store/prometheus.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import (
"encoding/json"
"fmt"
"io"
"math"
"net/http"
"net/url"
"path"
Expand Down Expand Up @@ -155,19 +156,24 @@ func (p *PrometheusStore) Series(r *storepb.SeriesRequest, s storepb.Store_Serie
continue
}

// We generally expect all samples of the requested range to be traversed
// so we just encode all samples into one big chunk regardless of size.
enc, cb, err := p.encodeChunk(e.Samples)
if err != nil {
return status.Error(codes.Unknown, err.Error())
sampleChunks := chunkSamples(e)

var aggregatedChunks []storepb.AggrChunk
for _, sampleChunk := range sampleChunks {
enc, cb, err := p.encodeChunk(sampleChunk)
if err != nil {
return status.Error(codes.Unknown, err.Error())
}
aggregatedChunks = append(aggregatedChunks, storepb.AggrChunk{
MinTime: int64(sampleChunk[0].Timestamp),
MaxTime: int64(sampleChunk[len(sampleChunk)-1].Timestamp),
Raw: &storepb.Chunk{Type: enc, Data: cb},
})
}

resp := storepb.NewSeriesResponse(&storepb.Series{
Labels: lset,
Chunks: []storepb.AggrChunk{{
MinTime: int64(e.Samples[0].Timestamp),
MaxTime: int64(e.Samples[len(e.Samples)-1].Timestamp),
Raw: &storepb.Chunk{Type: enc, Data: cb},
}},
Chunks: aggregatedChunks,
})
if err := s.Send(resp); err != nil {
return err
Expand All @@ -176,6 +182,23 @@ func (p *PrometheusStore) Series(r *storepb.SeriesRequest, s storepb.Store_Serie
return nil
}

// XoR encoding supports a max size of 2^16 - 1 samples, so we need
// to chunk all samples into groups of no more than 2^16 - 1
func chunkSamples(series prompb.TimeSeries) [][]prompb.Sample {
Copy link
Member

@bwplotka bwplotka Jan 9, 2019 •
edited
Loading

Choose a reason for hiding this comment

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

This name could be misleading as proper chunking should be with max 120 samples (:

I think we should name it bit differently for reader to explicitly state it is fine to do biggestPossible chunks in our query case. Alternatively name it like this but add maxSamples int64 argument.

Also we could consider, as we have lot's of raw samples here to decrease mem consumption for sidecar, to reuse e.Samples instead of recreating multiple of arrays of prombp.Sample. Each prompb.Samples array with lenght 2^16 is 1MB worth of mem. Maybe not much to optimize this not sure. Depends on how many samples at the end you have. Plus with lazy GC and multiple series it might be quite large number.

Up to you, we can start with something readable like this and move for microoptimizations later on or move to slicing and operate on indexes only e.g (I wrote it in bus, so I did not test this)

var aggr []storepb.AggrChunk
var arr []prompb.Sample = e.Samples // arr will share same underlying array
for len(arr) > 0 {
    len := len(arr)
    if i > math.MaxUint16  {
       len = math.MaxUint16
   }

    enc, cb, err := p.encodeChunk(arr[:len-1])
    if err != nil {
 	return status.Error(codes.Unknown, err.Error())
    }
    aggr = append(aggr, storepb.AggrChunk{
	MinTime: int64(arr[0].Timestamp),
	MaxTime: int64(arr[len-1].Timestamp),
	Raw:     &storepb.Chunk{Type: enc, Data: cb},
   })
   arr = arr[len-1:]
}

What do you think?

var sampleChunks [][]prompb.Sample
var currentSampleChunk []prompb.Sample
for i, sample := range series.Samples {
if i%math.MaxUint16 == 0 && i != 0 {
sampleChunks = append(sampleChunks, currentSampleChunk)
currentSampleChunk = []prompb.Sample{sample}
} else {
currentSampleChunk = append(currentSampleChunk, sample)
}
}
sampleChunks = append(sampleChunks, currentSampleChunk)
return sampleChunks
}

func (p *PrometheusStore) promSeries(ctx context.Context, q prompb.Query) (*prompb.ReadResponse, error) {
span, ctx := tracing.StartSpan(ctx, "query_prometheus")
defer span.Finish()
Expand Down
66 changes: 66 additions & 0 deletions pkg/store/prometheus_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package store
import (
"context"
"fmt"
"math"
"net/url"
"testing"
"time"
Expand Down Expand Up @@ -272,3 +273,68 @@ func TestPrometheusStore_Info(t *testing.T) {
testutil.Equals(t, int64(123), resp.MinTime)
testutil.Equals(t, int64(456), resp.MaxTime)
}

// Regression test for https://github.com/improbable-eng/thanos/issues/396.
func TestPrometheusStore_Series_SplitSamplesIntoChunksWithMaxSizeOfUint16_e2e(t *testing.T) {
defer leaktest.CheckTimeout(t, 10*time.Second)()
Copy link
Member

Choose a reason for hiding this comment

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

👍


p, err := testutil.NewPrometheus()
testutil.Ok(t, err)

baseT := timestamp.FromTime(time.Now().AddDate(0, 0, -2)) / 1000 * 1000

a := p.Appender()

offset := int64(math.MaxUint16 + 5)
Copy link
Member

Choose a reason for hiding this comment

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

Worth to make it 3 chunks maybe for solid test? (:

for i := int64(0); i < offset; i++ {
_, err = a.Add(labels.FromStrings("a", "b"), baseT + i, 1)
testutil.Ok(t, err)
}

testutil.Ok(t, a.Commit())

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

testutil.Ok(t, p.Start())
defer func() { testutil.Ok(t, p.Stop()) }()

u, err := url.Parse(fmt.Sprintf("http://%s", p.Addr()))
testutil.Ok(t, err)

proxy, err := NewPrometheusStore(nil, nil, u,
func() labels.Labels {
return labels.FromStrings("region", "eu-west")
}, nil)
testutil.Ok(t, err)
srv := newStoreSeriesServer(ctx)

err = proxy.Series(&storepb.SeriesRequest{
Copy link
Member

Choose a reason for hiding this comment

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

you can inline testutil.Ok(t, proxy.Series... but not a blocker.

MinTime: baseT,
MaxTime: baseT + offset,
Matchers: []storepb.LabelMatcher{
{Type: storepb.LabelMatcher_EQ, Name: "a", Value: "b"},
{Type: storepb.LabelMatcher_EQ, Name: "region", Value: "eu-west"},
},
}, srv)
testutil.Ok(t, err)

testutil.Equals(t, 1, len(srv.SeriesSet))

firstSeries := srv.SeriesSet[0]

testutil.Equals(t, []storepb.Label{
{Name: "a", Value: "b"},
{Name: "region", Value: "eu-west"},
}, firstSeries.Labels)

testutil.Equals(t, 2, len(firstSeries.Chunks))

chunk, err := chunkenc.FromData(chunkenc.EncXOR, firstSeries.Chunks[0].Raw.Data)
testutil.Ok(t, err)
testutil.Equals(t, math.MaxUint16, chunk.NumSamples())

chunk, err = chunkenc.FromData(chunkenc.EncXOR, firstSeries.Chunks[1].Raw.Data)
testutil.Ok(t, err)
testutil.Equals(t, 5, chunk.NumSamples())
}