Skip to content

Commit 2e693f2

Browse files
alanprotalanprot
authored
Fix Segmentation fault on ingester due race condition when reading blocks that are being deleted (#5119)
* fix query segfult Signed-off-by: Alan Protasio <alanprot@gmail.com> * Changelog Signed-off-by: Alan Protasio <alanprot@gmail.com> --------- Signed-off-by: Alan Protasio <alanprot@gmail.com>
1 parent 638ac6f commit 2e693f2

File tree

2 files changed

+115
-71
lines changed

2 files changed

+115
-71
lines changed

‎CHANGELOG.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232
* [BUGFIX] Fixed ingesters with less tokens stuck in LEAVING. #5061
3333
* [BUGFIX] Tracing: Fix missing object storage span instrumentation. #5074
3434
* [BUGFIX] Ingester: Ingesters returning empty response for metadata APIs. #5081
35+
* [BUGFIX] Ingester: Fix panic when querying metadata from blocks that are being deleted. #5119
3536
* [FEATURE] Alertmanager: Add support for time_intervals. #5102
3637

3738
## 1.14.0 2022-12-02

‎pkg/ingester/ingester.go

Lines changed: 114 additions & 71 deletions
Original file line numberDiff line numberDiff line change
@@ -1290,62 +1290,110 @@ func (i *Ingester) QueryExemplars(ctx context.Context, req *client.ExemplarQuery
12901290

12911291
// LabelValues returns all label values that are associated with a given label name.
12921292
func (i *Ingester) LabelValues(ctx context.Context, req *client.LabelValuesRequest) (*client.LabelValuesResponse, error) {
1293+
resp, cleanup, err := i.labelsValuesCommon(ctx, req)
1294+
defer cleanup()
1295+
return resp, err
1296+
}
1297+
1298+
// LabelValuesStream returns all label values that are associated with a given label name.
1299+
func (i *Ingester) LabelValuesStream(req *client.LabelValuesRequest, stream client.Ingester_LabelValuesStreamServer) error {
1300+
resp, cleanup, err := i.labelsValuesCommon(stream.Context(), req)
1301+
defer cleanup()
1302+
1303+
if err != nil {
1304+
return err
1305+
}
1306+
1307+
for i := 0; i < len(resp.LabelValues); i += metadataStreamBatchSize {
1308+
j := i + metadataStreamBatchSize
1309+
if j > len(resp.LabelValues) {
1310+
j = len(resp.LabelValues)
1311+
}
1312+
resp := &client.LabelValuesStreamResponse{
1313+
LabelValues: resp.LabelValues[i:j],
1314+
}
1315+
err := client.SendLabelValuesStream(stream, resp)
1316+
if err != nil {
1317+
return err
1318+
}
1319+
}
1320+
1321+
return nil
1322+
}
1323+
1324+
// labelsValuesCommon returns all label values that are associated with a given label name.
1325+
// this should be used by LabelValues and LabelValuesStream
1326+
// the cleanup function should be called in order to close the querier
1327+
func (i *Ingester) labelsValuesCommon(ctx context.Context, req *client.LabelValuesRequest) (*client.LabelValuesResponse, func(), error) {
1328+
cleanup := func() {}
12931329
if err := i.checkRunning(); err != nil {
1294-
return nil, err
1330+
return nil, cleanup, err
12951331
}
12961332

12971333
labelName, startTimestampMs, endTimestampMs, matchers, err := client.FromLabelValuesRequest(req)
12981334
if err != nil {
1299-
return nil, err
1335+
return nil, cleanup, err
13001336
}
13011337

13021338
userID, err := tenant.TenantID(ctx)
13031339
if err != nil {
1304-
return nil, err
1340+
return nil, cleanup, err
13051341
}
13061342

13071343
db := i.getTSDB(userID)
13081344
if db == nil {
1309-
return &client.LabelValuesResponse{}, nil
1345+
return &client.LabelValuesResponse{}, cleanup, nil
13101346
}
13111347

13121348
mint, maxt, err := metadataQueryRange(startTimestampMs, endTimestampMs, db, i.cfg.QueryStoreForLabels, i.cfg.QueryIngestersWithin)
13131349
if err != nil {
1314-
return nil, err
1350+
return nil, cleanup, err
13151351
}
13161352

13171353
q, err := db.Querier(ctx, mint, maxt)
13181354
if err != nil {
1319-
return nil, err
1355+
return nil, cleanup, err
1356+
}
1357+
1358+
cleanup = func() {
1359+
q.Close()
13201360
}
1321-
defer q.Close()
13221361

13231362
vals, _, err := q.LabelValues(labelName, matchers...)
13241363
if err != nil {
1325-
return nil, err
1364+
return nil, cleanup, err
13261365
}
13271366

13281367
return &client.LabelValuesResponse{
13291368
LabelValues: vals,
1330-
}, nil
1369+
}, cleanup, nil
13311370
}
13321371

1333-
func (i *Ingester) LabelValuesStream(req *client.LabelValuesRequest, stream client.Ingester_LabelValuesStreamServer) error {
1334-
resp, err := i.LabelValues(stream.Context(), req)
1372+
// LabelNames return all the label names.
1373+
func (i *Ingester) LabelNames(ctx context.Context, req *client.LabelNamesRequest) (*client.LabelNamesResponse, error) {
1374+
resp, cleanup, err := i.labelNamesCommon(ctx, req)
1375+
defer cleanup()
1376+
return resp, err
1377+
}
1378+
1379+
// LabelNamesStream return all the label names.
1380+
func (i *Ingester) LabelNamesStream(req *client.LabelNamesRequest, stream client.Ingester_LabelNamesStreamServer) error {
1381+
resp, cleanup, err := i.labelNamesCommon(stream.Context(), req)
1382+
defer cleanup()
13351383

13361384
if err != nil {
13371385
return err
13381386
}
13391387

1340-
for i := 0; i < len(resp.LabelValues); i += metadataStreamBatchSize {
1388+
for i := 0; i < len(resp.LabelNames); i += metadataStreamBatchSize {
13411389
j := i + metadataStreamBatchSize
1342-
if j > len(resp.LabelValues) {
1343-
j = len(resp.LabelValues)
1390+
if j > len(resp.LabelNames) {
1391+
j = len(resp.LabelNames)
13441392
}
1345-
resp := &client.LabelValuesStreamResponse{
1346-
LabelValues: resp.LabelValues[i:j],
1393+
resp := &client.LabelNamesStreamResponse{
1394+
LabelNames: resp.LabelNames[i:j],
13471395
}
1348-
err := client.SendLabelValuesStream(stream, resp)
1396+
err := client.SendLabelNamesStream(stream, resp)
13491397
if err != nil {
13501398
return err
13511399
}
@@ -1354,60 +1402,72 @@ func (i *Ingester) LabelValuesStream(req *client.LabelValuesRequest, stream clie
13541402
return nil
13551403
}
13561404

1357-
// LabelNames return all the label names.
1358-
func (i *Ingester) LabelNames(ctx context.Context, req *client.LabelNamesRequest) (*client.LabelNamesResponse, error) {
1405+
// labelNamesCommon return all the label names.
1406+
// this should be used by LabelNames and LabelNamesStream.
1407+
// the cleanup function should be called in order to close the querier
1408+
func (i *Ingester) labelNamesCommon(ctx context.Context, req *client.LabelNamesRequest) (*client.LabelNamesResponse, func(), error) {
1409+
cleanup := func() {}
13591410
if err := i.checkRunning(); err != nil {
1360-
return nil, err
1411+
return nil, cleanup, err
13611412
}
13621413

13631414
userID, err := tenant.TenantID(ctx)
13641415
if err != nil {
1365-
return nil, err
1416+
return nil, cleanup, err
13661417
}
13671418

13681419
db := i.getTSDB(userID)
13691420
if db == nil {
1370-
return &client.LabelNamesResponse{}, nil
1421+
return &client.LabelNamesResponse{}, cleanup, nil
13711422
}
13721423

13731424
mint, maxt, err := metadataQueryRange(req.StartTimestampMs, req.EndTimestampMs, db, i.cfg.QueryStoreForLabels, i.cfg.QueryIngestersWithin)
13741425
if err != nil {
1375-
return nil, err
1426+
return nil, cleanup, err
13761427
}
13771428

13781429
q, err := db.Querier(ctx, mint, maxt)
13791430
if err != nil {
1380-
return nil, err
1431+
return nil, cleanup, err
1432+
}
1433+
1434+
cleanup = func() {
1435+
q.Close()
13811436
}
1382-
defer q.Close()
13831437

13841438
names, _, err := q.LabelNames()
13851439
if err != nil {
1386-
return nil, err
1440+
return nil, cleanup, err
13871441
}
13881442

13891443
return &client.LabelNamesResponse{
13901444
LabelNames: names,
1391-
}, nil
1445+
}, cleanup, nil
13921446
}
13931447

1394-
// LabelNamesStream return all the label names.
1395-
func (i *Ingester) LabelNamesStream(req *client.LabelNamesRequest, stream client.Ingester_LabelNamesStreamServer) error {
1396-
resp, err := i.LabelNames(stream.Context(), req)
1448+
// MetricsForLabelMatchers returns all the metrics which match a set of matchers.
1449+
func (i *Ingester) MetricsForLabelMatchers(ctx context.Context, req *client.MetricsForLabelMatchersRequest) (*client.MetricsForLabelMatchersResponse, error) {
1450+
result, cleanup, err := i.metricsForLabelMatchersCommon(ctx, req)
1451+
defer cleanup()
1452+
return result, err
1453+
}
13971454

1455+
func (i *Ingester) MetricsForLabelMatchersStream(req *client.MetricsForLabelMatchersRequest, stream client.Ingester_MetricsForLabelMatchersStreamServer) error {
1456+
result, cleanup, err := i.metricsForLabelMatchersCommon(stream.Context(), req)
1457+
defer cleanup()
13981458
if err != nil {
13991459
return err
14001460
}
14011461

1402-
for i := 0; i < len(resp.LabelNames); i += metadataStreamBatchSize {
1462+
for i := 0; i < len(result.Metric); i += metadataStreamBatchSize {
14031463
j := i + metadataStreamBatchSize
1404-
if j > len(resp.LabelNames) {
1405-
j = len(resp.LabelNames)
1464+
if j > len(result.Metric) {
1465+
j = len(result.Metric)
14061466
}
1407-
resp := &client.LabelNamesStreamResponse{
1408-
LabelNames: resp.LabelNames[i:j],
1467+
resp := &client.MetricsForLabelMatchersStreamResponse{
1468+
Metric: result.Metric[i:j],
14091469
}
1410-
err := client.SendLabelNamesStream(stream, resp)
1470+
err := client.SendMetricsForLabelMatchersStream(stream, resp)
14111471
if err != nil {
14121472
return err
14131473
}
@@ -1416,46 +1476,52 @@ func (i *Ingester) LabelNamesStream(req *client.LabelNamesRequest, stream client
14161476
return nil
14171477
}
14181478

1419-
// MetricsForLabelMatchers returns all the metrics which match a set of matchers.
1420-
func (i *Ingester) MetricsForLabelMatchers(ctx context.Context, req *client.MetricsForLabelMatchersRequest) (*client.MetricsForLabelMatchersResponse, error) {
1479+
// metricsForLabelMatchersCommon returns all the metrics which match a set of matchers.
1480+
// this should be used by MetricsForLabelMatchers and MetricsForLabelMatchersStream.
1481+
// the cleanup function should be called in order to close the querier
1482+
func (i *Ingester) metricsForLabelMatchersCommon(ctx context.Context, req *client.MetricsForLabelMatchersRequest) (*client.MetricsForLabelMatchersResponse, func(), error) {
1483+
cleanup := func() {}
14211484
if err := i.checkRunning(); err != nil {
1422-
return nil, err
1485+
return nil, cleanup, err
14231486
}
14241487

14251488
userID, err := tenant.TenantID(ctx)
14261489
if err != nil {
1427-
return nil, err
1490+
return nil, cleanup, err
14281491
}
14291492

14301493
db := i.getTSDB(userID)
14311494
if db == nil {
1432-
return &client.MetricsForLabelMatchersResponse{}, nil
1495+
return &client.MetricsForLabelMatchersResponse{}, cleanup, nil
14331496
}
14341497

14351498
// Parse the request
14361499
_, _, matchersSet, err := client.FromMetricsForLabelMatchersRequest(req)
14371500
if err != nil {
1438-
return nil, err
1501+
return nil, cleanup, err
14391502
}
14401503

14411504
mint, maxt, err := metadataQueryRange(req.StartTimestampMs, req.EndTimestampMs, db, i.cfg.QueryStoreForLabels, i.cfg.QueryIngestersWithin)
14421505
if err != nil {
1443-
return nil, err
1506+
return nil, cleanup, err
14441507
}
14451508

14461509
q, err := db.Querier(ctx, mint, maxt)
14471510
if err != nil {
1448-
return nil, err
1511+
return nil, cleanup, err
1512+
}
1513+
1514+
cleanup = func() {
1515+
q.Close()
14491516
}
1450-
defer q.Close()
14511517

14521518
// Run a query for each matchers set and collect all the results.
14531519
var sets []storage.SeriesSet
14541520

14551521
for _, matchers := range matchersSet {
14561522
// Interrupt if the context has been canceled.
14571523
if ctx.Err() != nil {
1458-
return nil, ctx.Err()
1524+
return nil, cleanup, ctx.Err()
14591525
}
14601526

14611527
hints := &storage.SelectHints{
@@ -1477,38 +1543,15 @@ func (i *Ingester) MetricsForLabelMatchers(ctx context.Context, req *client.Metr
14771543
for mergedSet.Next() {
14781544
// Interrupt if the context has been canceled.
14791545
if ctx.Err() != nil {
1480-
return nil, ctx.Err()
1546+
return nil, cleanup, ctx.Err()
14811547
}
14821548

14831549
result.Metric = append(result.Metric, &cortexpb.Metric{
14841550
Labels: cortexpb.FromLabelsToLabelAdapters(mergedSet.At().Labels()),
14851551
})
14861552
}
14871553

1488-
return result, nil
1489-
}
1490-
1491-
func (i *Ingester) MetricsForLabelMatchersStream(req *client.MetricsForLabelMatchersRequest, stream client.Ingester_MetricsForLabelMatchersStreamServer) error {
1492-
result, err := i.MetricsForLabelMatchers(stream.Context(), req)
1493-
if err != nil {
1494-
return err
1495-
}
1496-
1497-
for i := 0; i < len(result.Metric); i += metadataStreamBatchSize {
1498-
j := i + metadataStreamBatchSize
1499-
if j > len(result.Metric) {
1500-
j = len(result.Metric)
1501-
}
1502-
resp := &client.MetricsForLabelMatchersStreamResponse{
1503-
Metric: result.Metric[i:j],
1504-
}
1505-
err := client.SendMetricsForLabelMatchersStream(stream, resp)
1506-
if err != nil {
1507-
return err
1508-
}
1509-
}
1510-
1511-
return nil
1554+
return result, cleanup, nil
15121555
}
15131556

15141557
// MetricsMetadata returns all the metric metadata of a user.

0 commit comments

Comments
 (0)