Skip to content

Commit 80124e3

Browse files
SungJin1212SungJin1212
committed
Add prealloc timeseies v2
Signed-off-by: SungJin1212 <[email protected]>
1 parent 878562f commit 80124e3

17 files changed

+501
-235
lines changed

‎integration/e2e/util.go

Lines changed: 11 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,6 @@ import (
2525
"github.com/thanos-io/thanos/pkg/block/metadata"
2626
"github.com/thanos-io/thanos/pkg/runutil"
2727

28-
"github.com/cortexproject/cortex/pkg/cortexpbv2"
2928
cortex_tsdb "github.com/cortexproject/cortex/pkg/storage/tsdb"
3029
)
3130

@@ -151,7 +150,7 @@ func GenerateSeries(name string, ts time.Time, additionalLabels ...prompb.Label)
151150
return
152151
}
153152

154-
func GenerateHistogramSeriesV2(name string, ts time.Time, i uint32, floatHistogram bool, additionalLabels ...prompb.Label) (symbols []string, series []cortexpbv2.TimeSeries) {
153+
func GenerateHistogramSeriesV2(name string, ts time.Time, i uint32, floatHistogram bool, additionalLabels ...prompb.Label) (symbols []string, series []writev2.TimeSeries) {
155154
tsMillis := TimeToMilliseconds(ts)
156155

157156
st := writev2.NewSymbolTable()
@@ -164,20 +163,20 @@ func GenerateHistogramSeriesV2(name string, ts time.Time, i uint32, floatHistogr
164163
var (
165164
h *histogram.Histogram
166165
fh *histogram.FloatHistogram
167-
ph cortexpbv2.Histogram
166+
ph writev2.Histogram
168167
)
169168
if floatHistogram {
170169
fh = tsdbutil.GenerateTestFloatHistogram(int(i))
171-
ph = cortexpbv2.FloatHistogramToHistogramProto(tsMillis, fh)
170+
ph = writev2.FromFloatHistogram(tsMillis, fh)
172171
} else {
173172
h = tsdbutil.GenerateTestHistogram(int(i))
174-
ph = cortexpbv2.HistogramToHistogramProto(tsMillis, h)
173+
ph = writev2.FromIntHistogram(tsMillis, h)
175174
}
176175

177176
// Generate the series
178-
series = append(series, cortexpbv2.TimeSeries{
177+
series = append(series, writev2.TimeSeries{
179178
LabelsRefs: st.SymbolizeLabels(lbs, nil),
180-
Histograms: []cortexpbv2.Histogram{ph},
179+
Histograms: []writev2.Histogram{ph},
181180
})
182181

183182
symbols = st.Symbols()
@@ -224,7 +223,7 @@ func GenerateHistogramSeries(name string, ts time.Time, i uint32, floatHistogram
224223
return
225224
}
226225

227-
func GenerateSeriesV2(name string, ts time.Time, additionalLabels ...prompb.Label) (symbols []string, series []cortexpbv2.TimeSeries, vector model.Vector) {
226+
func GenerateSeriesV2(name string, ts time.Time, additionalLabels ...prompb.Label) (symbols []string, series []writev2.TimeSeries, vector model.Vector) {
228227
tsMillis := TimeToMilliseconds(ts)
229228
value := rand.Float64()
230229

@@ -237,14 +236,14 @@ func GenerateSeriesV2(name string, ts time.Time, additionalLabels ...prompb.Labe
237236
Value: label.Value,
238237
})
239238
}
240-
series = append(series, cortexpbv2.TimeSeries{
239+
series = append(series, writev2.TimeSeries{
241240
// Generate the series
242241
LabelsRefs: st.SymbolizeLabels(lbs, nil),
243-
Samples: []cortexpbv2.Sample{
242+
Samples: []writev2.Sample{
244243
{Value: value, Timestamp: tsMillis},
245244
},
246-
Metadata: cortexpbv2.Metadata{
247-
Type: cortexpbv2.METRIC_TYPE_GAUGE,
245+
Metadata: writev2.Metadata{
246+
Type: writev2.Metadata_METRIC_TYPE_GAUGE,
248247
},
249248
})
250249
symbols = st.Symbols()

‎integration/e2ecortex/client.go

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -23,14 +23,14 @@ import (
2323
"github.com/prometheus/prometheus/model/labels"
2424
"github.com/prometheus/prometheus/model/rulefmt"
2525
"github.com/prometheus/prometheus/prompb"
26+
writev2 "github.com/prometheus/prometheus/prompb/io/prometheus/write/v2"
2627
"github.com/prometheus/prometheus/storage"
2728
"github.com/prometheus/prometheus/storage/remote"
2829
"go.opentelemetry.io/collector/pdata/pcommon"
2930
"go.opentelemetry.io/collector/pdata/pmetric"
3031
"go.opentelemetry.io/collector/pdata/pmetric/pmetricotlp"
3132
yaml "gopkg.in/yaml.v3"
3233

33-
"github.com/cortexproject/cortex/pkg/cortexpbv2"
3434
"github.com/cortexproject/cortex/pkg/ruler"
3535
"github.com/cortexproject/cortex/pkg/util/backoff"
3636
)
@@ -114,9 +114,9 @@ func NewPromQueryClient(address string) (*Client, error) {
114114
}
115115

116116
// PushV2 the input timeseries to the remote endpoint
117-
func (c *Client) PushV2(symbols []string, timeseries []cortexpbv2.TimeSeries) (*http.Response, error) {
117+
func (c *Client) PushV2(symbols []string, timeseries []writev2.TimeSeries) (*http.Response, error) {
118118
// Create write request
119-
data, err := proto.Marshal(&cortexpbv2.WriteRequest{Symbols: symbols, Timeseries: timeseries})
119+
data, err := proto.Marshal(&writev2.Request{Symbols: symbols, Timeseries: timeseries})
120120
if err != nil {
121121
return nil, err
122122
}

‎integration/remote_write_v2_test.go

Lines changed: 13 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -12,14 +12,14 @@ import (
1212

1313
"github.com/prometheus/common/model"
1414
"github.com/prometheus/prometheus/prompb"
15+
writev2 "github.com/prometheus/prometheus/prompb/io/prometheus/write/v2"
1516
"github.com/prometheus/prometheus/tsdb/tsdbutil"
1617
"github.com/stretchr/testify/assert"
1718
"github.com/stretchr/testify/require"
1819

1920
"github.com/cortexproject/cortex/integration/e2e"
2021
e2edb "github.com/cortexproject/cortex/integration/e2e/db"
2122
"github.com/cortexproject/cortex/integration/e2ecortex"
22-
"github.com/cortexproject/cortex/pkg/cortexpbv2"
2323
"github.com/cortexproject/cortex/pkg/storage/tsdb"
2424
)
2525

@@ -176,24 +176,22 @@ func TestExemplar(t *testing.T) {
176176
now := time.Now()
177177
tsMillis := e2e.TimeToMilliseconds(now)
178178

179-
req := &cortexpbv2.WriteRequest{
180-
Symbols: []string{"", "__name__", "test_metric", "b", "c", "baz", "qux", "d", "e", "foo", "bar", "f", "g", "h", "i", "Test gauge for test purposes", "Maybe op/sec who knows (:", "Test counter for test purposes"},
181-
Timeseries: []cortexpbv2.TimeSeries{
182-
{
183-
LabelsRefs: []uint32{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, // Symbolized writeRequestFixture.Timeseries[0].Labels
184-
Metadata: cortexpbv2.Metadata{
185-
Type: cortexpbv2.METRIC_TYPE_COUNTER, // writeV2RequestSeries1Metadata.Type.
186-
187-
HelpRef: 15, // Symbolized writeV2RequestSeries1Metadata.Help.
188-
UnitRef: 16, // Symbolized writeV2RequestSeries1Metadata.Unit.
189-
},
190-
Samples: []cortexpbv2.Sample{{Value: 1, Timestamp: tsMillis}},
191-
Exemplars: []cortexpbv2.Exemplar{{LabelsRefs: []uint32{11, 12}, Value: 1, Timestamp: tsMillis}},
179+
symbols := []string{"", "__name__", "test_metric", "b", "c", "baz", "qux", "d", "e", "foo", "bar", "f", "g", "h", "i", "Test gauge for test purposes", "Maybe op/sec who knows (:", "Test counter for test purposes"}
180+
timeseries := []writev2.TimeSeries{
181+
{
182+
LabelsRefs: []uint32{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, // Symbolized writeRequestFixture.Timeseries[0].Labels
183+
Metadata: writev2.Metadata{
184+
Type: writev2.Metadata_METRIC_TYPE_COUNTER, // writeV2RequestSeries1Metadata.Type.
185+
186+
HelpRef: 15, // Symbolized writeV2RequestSeries1Metadata.Help.
187+
UnitRef: 16, // Symbolized writeV2RequestSeries1Metadata.Unit.
192188
},
189+
Samples: []writev2.Sample{{Value: 1, Timestamp: tsMillis}},
190+
Exemplars: []writev2.Exemplar{{LabelsRefs: []uint32{11, 12}, Value: 1, Timestamp: tsMillis}},
193191
},
194192
}
195193

196-
res, err := c.PushV2(req.Symbols, req.Timeseries)
194+
res, err := c.PushV2(symbols, timeseries)
197195
require.NoError(t, err)
198196
require.Equal(t, 200, res.StatusCode)
199197
testPushHeader(t, res.Header, "1", "0", "1")

‎pkg/cortexpb/slicesPool.go

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@ type byteSlicePools struct {
1313
pools []sync.Pool
1414
}
1515

16-
func newSlicePool(pools int) *byteSlicePools {
16+
func NewSlicePool(pools int) *byteSlicePools {
1717
sp := byteSlicePools{}
1818
sp.init(pools)
1919
return &sp
@@ -32,7 +32,7 @@ func (sp *byteSlicePools) init(pools int) {
3232
}
3333
}
3434

35-
func (sp *byteSlicePools) getSlice(size int) *[]byte {
35+
func (sp *byteSlicePools) GetSlice(size int) *[]byte {
3636
index := int(math.Ceil(math.Log2(float64(size)))) - minPoolSizePower
3737

3838
if index >= len(sp.pools) {
@@ -50,7 +50,7 @@ func (sp *byteSlicePools) getSlice(size int) *[]byte {
5050
return s
5151
}
5252

53-
func (sp *byteSlicePools) reuseSlice(s *[]byte) {
53+
func (sp *byteSlicePools) ReuseSlice(s *[]byte) {
5454
index := int(math.Floor(math.Log2(float64(cap(*s))))) - minPoolSizePower
5555

5656
if index >= len(sp.pools) || index < 0 {

‎pkg/cortexpb/slicesPool_test.go

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -9,22 +9,22 @@ import (
99
)
1010

1111
func TestFuzzyByteSlicePools(t *testing.T) {
12-
sut := newSlicePool(20)
12+
sut := NewSlicePool(20)
1313
maxByteSize := int(math.Pow(2, 20+minPoolSizePower-1))
1414

1515
for i := 0; i < 1000; i++ {
1616
size := rand.Int() % maxByteSize
17-
s := sut.getSlice(size)
17+
s := sut.GetSlice(size)
1818
assert.Equal(t, len(*s), size)
19-
sut.reuseSlice(s)
19+
sut.ReuseSlice(s)
2020
}
2121
}
2222

2323
func TestReturnSliceSmallerThanMin(t *testing.T) {
24-
sut := newSlicePool(20)
24+
sut := NewSlicePool(20)
2525
size := 3
2626
buff := make([]byte, 0, size)
27-
sut.reuseSlice(&buff)
28-
buff2 := sut.getSlice(size * 2)
27+
sut.ReuseSlice(&buff)
28+
buff2 := sut.GetSlice(size * 2)
2929
assert.Equal(t, len(*buff2), size*2)
3030
}

‎pkg/cortexpb/timeseries.go

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,7 @@ var (
4747
}
4848
},
4949
}
50-
bytePool = newSlicePool(20)
50+
bytePool = NewSlicePool(20)
5151
)
5252

5353
// PreallocConfig configures how structures will be preallocated to optimise
@@ -86,7 +86,7 @@ func (p *PreallocTimeseries) Unmarshal(dAtA []byte) error {
8686

8787
func (p *PreallocWriteRequest) Marshal() (dAtA []byte, err error) {
8888
size := p.Size()
89-
p.data = bytePool.getSlice(size)
89+
p.data = bytePool.GetSlice(size)
9090
dAtA = *p.data
9191
n, err := p.MarshalToSizedBuffer(dAtA[:size])
9292
if err != nil {
@@ -97,7 +97,7 @@ func (p *PreallocWriteRequest) Marshal() (dAtA []byte, err error) {
9797

9898
func ReuseWriteRequest(req *PreallocWriteRequest) {
9999
if req.data != nil {
100-
bytePool.reuseSlice(req.data)
100+
bytePool.ReuseSlice(req.data)
101101
req.data = nil
102102
}
103103
req.Source = 0

‎pkg/cortexpbv2/compatv2.go

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -22,13 +22,14 @@ func ToWriteRequestV2(lbls []labels.Labels, samples []Sample, histograms []Histo
2222
symbols := st.Symbols()
2323

2424
req := &WriteRequest{
25-
Symbols: symbols,
26-
Source: source,
25+
Timeseries: PreallocTimeseriesV2SliceFromPool(),
26+
Symbols: symbols,
27+
Source: source,
2728
}
2829

2930
i := 0
3031
for i < len(samples) || i < len(histograms) || i < len(metadata) {
31-
ts := TimeSeries{}
32+
ts := TimeseriesV2FromPool()
3233
ts.LabelsRefs = labelRefs[i]
3334
if i < len(samples) {
3435
ts.Samples = append(ts.Samples, samples[i])
@@ -40,7 +41,7 @@ func ToWriteRequestV2(lbls []labels.Labels, samples []Sample, histograms []Histo
4041
ts.Metadata = metadata[i]
4142
}
4243
i++
43-
req.Timeseries = append(req.Timeseries, ts)
44+
req.Timeseries = append(req.Timeseries, PreallocTimeseriesV2{TimeSeries: ts})
4445
}
4546

4647
return req

0 commit comments

Comments
 (0)