Skip to content

Commit a981316

Browse files
dinalGal TopperDina NimroditalIguaz
authored
Development (#477)
* remove check if num query workers is a power of 2 (#474) * remove check if num query workers is a power of 2 * remove numWorkers check Co-authored-by: Dina Nimrodi <[email protected]> * IG-15260 (#476) * query 2 last partitions when querying for all labelsets * add test, minor bug * fix comment Co-authored-by: Gal Topper <[email protected]> Co-authored-by: Dina Nimrodi <[email protected]> Co-authored-by: Tal Neiman <[email protected]>
1 parent e2513db commit a981316

File tree

3 files changed

+91
-35
lines changed

3 files changed

+91
-35
lines changed

Diff for: ‎pkg/pquerier/pqueriertest/get_labelsets_integration_test.go

+40
Original file line numberDiff line numberDiff line change
@@ -206,3 +206,43 @@ func (suite *getLabelSetsSuite) TestGetLabelsWithFilter() {
206206

207207
suite.ElementsMatch(expectedLabels, labelsList, "actual label sets does not match expected")
208208
}
209+
210+
func (suite *getLabelSetsSuite) TestGetLabelsAllMetricsFrom2Partitions() {
211+
adapter, err := tsdb.NewV3ioAdapter(suite.v3ioConfig, nil, nil)
212+
suite.Require().NoError(err, "failed to create v3io adapter")
213+
214+
labels := []utils.Labels{utils.LabelsFromStringList("os", "linux", "region", "europe"),
215+
utils.LabelsFromStringList("os", "linux", "region", "asia"),
216+
utils.LabelsFromStringList("os", "mac", "region", "europe")}
217+
testParams := tsdbtest.NewTestParams(suite.T(),
218+
tsdbtest.TestOption{
219+
Key: tsdbtest.OptTimeSeries,
220+
Value: tsdbtest.TimeSeries{tsdbtest.Metric{
221+
Name: "cpu",
222+
Labels: labels[0],
223+
Data: []tsdbtest.DataPoint{{suite.basicQueryTime - 4*tsdbtest.DaysInMillis, 10}}},
224+
tsdbtest.Metric{
225+
Name: "cpu",
226+
Labels: labels[1],
227+
Data: []tsdbtest.DataPoint{{suite.basicQueryTime - 4*tsdbtest.DaysInMillis, 10},
228+
{suite.basicQueryTime - 2*tsdbtest.DaysInMillis, 10}}},
229+
tsdbtest.Metric{
230+
Name: "cpu",
231+
Labels: labels[2],
232+
Data: []tsdbtest.DataPoint{{suite.basicQueryTime, 10}}},
233+
}})
234+
tsdbtest.InsertData(suite.T(), testParams)
235+
expectedLabels := []utils.Labels{
236+
utils.LabelsFromStringList("os", "linux", "region", "asia", config.PrometheusMetricNameAttribute, "cpu"),
237+
utils.LabelsFromStringList("os", "mac", "region", "europe", config.PrometheusMetricNameAttribute, "cpu")}
238+
239+
querierV2, err := adapter.QuerierV2()
240+
suite.Require().NoError(err, "failed to create querier v2")
241+
242+
labelsList, err := querierV2.GetLabelSets("", "")
243+
if err != nil {
244+
suite.T().Fatalf("failed to get label sets, err:%v\n", err)
245+
}
246+
247+
suite.Require().ElementsMatch(expectedLabels, labelsList, "actual label sets does not match expected")
248+
}

Diff for: ‎pkg/pquerier/querier.go

+6-2
Original file line numberDiff line numberDiff line change
@@ -321,12 +321,16 @@ func (q *V3ioQuerier) GetLabelSets(metric string, filter string) ([]utils.Labels
321321

322322
// Get all label sets
323323
input := v3io.GetItemsInput{
324-
Path: partitionPaths[0],
325324
Filter: filter,
326325
AttributeNames: []string{config.LabelSetAttrName, config.MetricNameAttrName},
327326
}
328327

329-
iter, err := utils.NewAsyncItemsCursor(q.container, &input, q.cfg.QryWorkers, shardingKeys, q.logger)
328+
// Because of performance issues we only want to query the last two partitions
329+
partitionsToQuery := []string{partitionPaths[len(partitionPaths)-1]}
330+
if len(partitionPaths) > 1 {
331+
partitionsToQuery = append(partitionsToQuery, partitionPaths[len(partitionPaths)-2])
332+
}
333+
iter, err := utils.NewAsyncItemsCursorMultiplePartitions(q.container, &input, q.cfg.QryWorkers, shardingKeys, q.logger, partitionsToQuery)
330334
if err != nil {
331335
return nil, err
332336
}

Diff for: ‎pkg/utils/asynciter.go

+45-33
Original file line numberDiff line numberDiff line change
@@ -48,68 +48,80 @@ type AsyncItemsCursor struct {
4848
container v3io.Container
4949
logger logger.Logger
5050

51-
responseChan chan *v3io.Response
52-
workers int
53-
totalSegments int
54-
lastShards int
55-
Cnt int
51+
responseChan chan *v3io.Response
52+
workers int
53+
totalSegments int
54+
lastShards int
55+
Cnt int
56+
numberOfPartitions int
5657
}
5758

58-
func NewAsyncItemsCursor(container v3io.Container, input *v3io.GetItemsInput, workers int, shardingKeys []string, logger logger.Logger) (*AsyncItemsCursor, error) {
59+
func NewAsyncItemsCursorMultiplePartitions(container v3io.Container, input *v3io.GetItemsInput, workers int,
60+
shardingKeys []string, logger logger.Logger, partitions []string) (*AsyncItemsCursor, error) {
5961

6062
// TODO: use workers from Context.numWorkers (if no ShardingKey)
6163
if workers == 0 || input.ShardingKey != "" {
6264
workers = 1
6365
}
6466

6567
newAsyncItemsCursor := &AsyncItemsCursor{
66-
container: container,
67-
input: input,
68-
responseChan: make(chan *v3io.Response, 1000),
69-
workers: workers,
70-
logger: logger.GetChild("AsyncItemsCursor"),
68+
container: container,
69+
input: input,
70+
responseChan: make(chan *v3io.Response, 1000),
71+
workers: workers,
72+
logger: logger.GetChild("AsyncItemsCursor"),
73+
numberOfPartitions: len(partitions),
7174
}
7275

7376
if len(shardingKeys) > 0 {
7477
newAsyncItemsCursor.workers = len(shardingKeys)
7578

79+
for _, partition := range partitions {
80+
for i := 0; i < newAsyncItemsCursor.workers; i++ {
81+
input := v3io.GetItemsInput{
82+
Path: partition,
83+
AttributeNames: input.AttributeNames,
84+
Filter: input.Filter,
85+
ShardingKey: shardingKeys[i],
86+
}
87+
_, err := container.GetItems(&input, &input, newAsyncItemsCursor.responseChan)
88+
89+
if err != nil {
90+
return nil, err
91+
}
92+
}
93+
}
94+
95+
return newAsyncItemsCursor, nil
96+
}
97+
98+
for _, partition := range partitions {
7699
for i := 0; i < newAsyncItemsCursor.workers; i++ {
100+
newAsyncItemsCursor.totalSegments = workers
77101
input := v3io.GetItemsInput{
78-
Path: input.Path,
102+
Path: partition,
79103
AttributeNames: input.AttributeNames,
80104
Filter: input.Filter,
81-
ShardingKey: shardingKeys[i],
105+
TotalSegments: newAsyncItemsCursor.totalSegments,
106+
Segment: i,
82107
}
83108
_, err := container.GetItems(&input, &input, newAsyncItemsCursor.responseChan)
84109

85110
if err != nil {
111+
// TODO: proper exit, release requests which passed
86112
return nil, err
87113
}
88114
}
89-
90-
return newAsyncItemsCursor, nil
91-
}
92-
93-
for i := 0; i < newAsyncItemsCursor.workers; i++ {
94-
newAsyncItemsCursor.totalSegments = workers
95-
input := v3io.GetItemsInput{
96-
Path: input.Path,
97-
AttributeNames: input.AttributeNames,
98-
Filter: input.Filter,
99-
TotalSegments: newAsyncItemsCursor.totalSegments,
100-
Segment: i,
101-
}
102-
_, err := container.GetItems(&input, &input, newAsyncItemsCursor.responseChan)
103-
104-
if err != nil {
105-
// TODO: proper exit, release requests which passed
106-
return nil, err
107-
}
108115
}
109116

110117
return newAsyncItemsCursor, nil
111118
}
112119

120+
func NewAsyncItemsCursor(container v3io.Container, input *v3io.GetItemsInput, workers int, shardingKeys []string,
121+
logger logger.Logger) (*AsyncItemsCursor, error) {
122+
return NewAsyncItemsCursorMultiplePartitions(container, input, workers, shardingKeys, logger, []string{input.Path})
123+
}
124+
113125
// error returns the last error
114126
func (ic *AsyncItemsCursor) Err() error {
115127
return ic.currentError
@@ -148,7 +160,7 @@ func (ic *AsyncItemsCursor) NextItem() (v3io.Item, error) {
148160
}
149161

150162
// are there any more items up stream? did all the shards complete ?
151-
if ic.lastShards == ic.workers {
163+
if ic.lastShards == ic.workers*ic.numberOfPartitions {
152164
ic.currentError = nil
153165
return nil, nil
154166
}

0 commit comments

Comments
 (0)