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

Apply bytes limiter to LabelNames and LabelValuesForLabelNames #6568

Merged
merged 6 commits into from
Feb 3, 2025
Merged

Apply bytes limiter to LabelNames and LabelValuesForLabelNames #6568

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
c0acf9f
Add bytes limiter to LabelValuesForLabelNames
justinjung04 Jan 31, 2025
fe7c1f9
Add bytes limiter to LabelNames
justinjung04 Jan 31, 2025
4f0468d
Fix bug in MetricsForLabelMatchersStream where bytes limiter overwrot…
justinjung04 Jan 31, 2025
749068a
changelog
justinjung04 Jan 31, 2025
21edbf5
Apply bytes limiter to LabelNames and LabelValues in block store quer…
justinjung04 Jan 31, 2025
1d4f155
Merge branch 'master' into labels-bytes-limit
yeya24 Feb 3, 2025
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
1 change: 1 addition & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
## master / unreleased

* [FEATURE] Querier/Ruler: Add `query_partial_data` and `rules_partial_data` limits to allow queries/rules to be evaluated with data from a single zone, if other zones are not available. #6526
* [ENHANCEMENT] Querier: Apply bytes limiter to LabelNames and LabelValuesForLabelNames. #6568
* [ENHANCEMENT] Query Frontend: Add a `too_many_tenants` reason label value to `cortex_rejected_queries_total` metric to track the rejected query count due to the # of tenant limits. #6569
* [BUGFIX] Ingester: Avoid error or early throttling when READONLY ingesters are present in the ring #6517

Expand Down
43 changes: 29 additions & 14 deletions pkg/distributor/distributor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1188,7 +1188,7 @@ func (d *Distributor) ForReplicationSet(ctx context.Context, replicationSet ring
})
}

func (d *Distributor) LabelValuesForLabelNameCommon(ctx context.Context, from, to model.Time, labelName model.LabelName, hints *storage.LabelHints, f func(ctx context.Context, rs ring.ReplicationSet, req *ingester_client.LabelValuesRequest) ([]interface{}, error), matchers ...*labels.Matcher) ([]string, error) {
func (d *Distributor) LabelValuesForLabelNameCommon(ctx context.Context, from, to model.Time, labelName model.LabelName, hints *storage.LabelHints, f func(ctx context.Context, rs ring.ReplicationSet, req *ingester_client.LabelValuesRequest, limiter *limiter.QueryLimiter) ([]interface{}, error), matchers ...*labels.Matcher) ([]string, error) {
justinjung04 marked this conversation as resolved.
Show resolved Hide resolved
span, ctx := opentracing.StartSpanFromContext(ctx, "Distributor.LabelValues", opentracing.Tags{
"name": labelName,
"start": from.Unix(),
Expand All @@ -1205,7 +1205,8 @@ func (d *Distributor) LabelValuesForLabelNameCommon(ctx context.Context, from, t
return nil, err
}

resps, err := f(ctx, replicationSet, req)
queryLimiter := limiter.QueryLimiterFromContextWithFallback(ctx)
resps, err := f(ctx, replicationSet, req, queryLimiter)
if err != nil {
return nil, err
}
Expand All @@ -1229,20 +1230,23 @@ func (d *Distributor) LabelValuesForLabelNameCommon(ctx context.Context, from, t

// LabelValuesForLabelName returns all the label values that are associated with a given label name.
func (d *Distributor) LabelValuesForLabelName(ctx context.Context, from, to model.Time, labelName model.LabelName, hint *storage.LabelHints, partialDataEnabled bool, matchers ...*labels.Matcher) ([]string, error) {
return d.LabelValuesForLabelNameCommon(ctx, from, to, labelName, hint, func(ctx context.Context, rs ring.ReplicationSet, req *ingester_client.LabelValuesRequest) ([]interface{}, error) {
return d.LabelValuesForLabelNameCommon(ctx, from, to, labelName, hint, func(ctx context.Context, rs ring.ReplicationSet, req *ingester_client.LabelValuesRequest, queryLimiter *limiter.QueryLimiter) ([]interface{}, error) {
return d.ForReplicationSet(ctx, rs, d.cfg.ZoneResultsQuorumMetadata, partialDataEnabled, func(ctx context.Context, client ingester_client.IngesterClient) (interface{}, error) {
resp, err := client.LabelValues(ctx, req)
if err != nil {
return nil, err
}
if err := queryLimiter.AddDataBytes(resp.Size()); err != nil {
return nil, validation.LimitError(err.Error())
}
return resp.LabelValues, nil
})
}, matchers...)
}

// LabelValuesForLabelNameStream returns all the label values that are associated with a given label name.
func (d *Distributor) LabelValuesForLabelNameStream(ctx context.Context, from, to model.Time, labelName model.LabelName, hint *storage.LabelHints, partialDataEnabled bool, matchers ...*labels.Matcher) ([]string, error) {
return d.LabelValuesForLabelNameCommon(ctx, from, to, labelName, hint, func(ctx context.Context, rs ring.ReplicationSet, req *ingester_client.LabelValuesRequest) ([]interface{}, error) {
return d.LabelValuesForLabelNameCommon(ctx, from, to, labelName, hint, func(ctx context.Context, rs ring.ReplicationSet, req *ingester_client.LabelValuesRequest, queryLimiter *limiter.QueryLimiter) ([]interface{}, error) {
return d.ForReplicationSet(ctx, rs, d.cfg.ZoneResultsQuorumMetadata, partialDataEnabled, func(ctx context.Context, client ingester_client.IngesterClient) (interface{}, error) {
stream, err := client.LabelValuesStream(ctx, req)
if err != nil {
Expand All @@ -1252,12 +1256,15 @@ func (d *Distributor) LabelValuesForLabelNameStream(ctx context.Context, from, t
allLabelValues := []string{}
for {
resp, err := stream.Recv()

if err == io.EOF {
break
} else if err != nil {
return nil, err
}
if err := queryLimiter.AddDataBytes(resp.Size()); err != nil {
return nil, validation.LimitError(err.Error())
}

allLabelValues = append(allLabelValues, resp.LabelValues...)
}

Expand All @@ -1266,7 +1273,7 @@ func (d *Distributor) LabelValuesForLabelNameStream(ctx context.Context, from, t
}, matchers...)
}

func (d *Distributor) LabelNamesCommon(ctx context.Context, from, to model.Time, hints *storage.LabelHints, f func(ctx context.Context, rs ring.ReplicationSet, req *ingester_client.LabelNamesRequest) ([]interface{}, error), matchers ...*labels.Matcher) ([]string, error) {
func (d *Distributor) LabelNamesCommon(ctx context.Context, from, to model.Time, hints *storage.LabelHints, f func(ctx context.Context, rs ring.ReplicationSet, req *ingester_client.LabelNamesRequest, limiter *limiter.QueryLimiter) ([]interface{}, error), matchers ...*labels.Matcher) ([]string, error) {
span, ctx := opentracing.StartSpanFromContext(ctx, "Distributor.LabelNames", opentracing.Tags{
"start": from.Unix(),
"end": to.Unix(),
Expand All @@ -1283,7 +1290,8 @@ func (d *Distributor) LabelNamesCommon(ctx context.Context, from, to model.Time,
return nil, err
}

resps, err := f(ctx, replicationSet, req)
queryLimiter := limiter.QueryLimiterFromContextWithFallback(ctx)
resps, err := f(ctx, replicationSet, req, queryLimiter)
if err != nil {
return nil, err
}
Expand All @@ -1308,7 +1316,7 @@ func (d *Distributor) LabelNamesCommon(ctx context.Context, from, to model.Time,
}

func (d *Distributor) LabelNamesStream(ctx context.Context, from, to model.Time, hints *storage.LabelHints, partialDataEnabled bool, matchers ...*labels.Matcher) ([]string, error) {
return d.LabelNamesCommon(ctx, from, to, hints, func(ctx context.Context, rs ring.ReplicationSet, req *ingester_client.LabelNamesRequest) ([]interface{}, error) {
return d.LabelNamesCommon(ctx, from, to, hints, func(ctx context.Context, rs ring.ReplicationSet, req *ingester_client.LabelNamesRequest, queryLimiter *limiter.QueryLimiter) ([]interface{}, error) {
return d.ForReplicationSet(ctx, rs, d.cfg.ZoneResultsQuorumMetadata, partialDataEnabled, func(ctx context.Context, client ingester_client.IngesterClient) (interface{}, error) {
stream, err := client.LabelNamesStream(ctx, req)
if err != nil {
Expand All @@ -1318,12 +1326,15 @@ func (d *Distributor) LabelNamesStream(ctx context.Context, from, to model.Time,
allLabelNames := []string{}
for {
resp, err := stream.Recv()

if err == io.EOF {
break
} else if err != nil {
return nil, err
}
if err := queryLimiter.AddDataBytes(resp.Size()); err != nil {
return nil, validation.LimitError(err.Error())
}

allLabelNames = append(allLabelNames, resp.LabelNames...)
}

Expand All @@ -1334,12 +1345,16 @@ func (d *Distributor) LabelNamesStream(ctx context.Context, from, to model.Time,

// LabelNames returns all the label names.
func (d *Distributor) LabelNames(ctx context.Context, from, to model.Time, hint *storage.LabelHints, partialDataEnabled bool, matchers ...*labels.Matcher) ([]string, error) {
return d.LabelNamesCommon(ctx, from, to, hint, func(ctx context.Context, rs ring.ReplicationSet, req *ingester_client.LabelNamesRequest) ([]interface{}, error) {
return d.LabelNamesCommon(ctx, from, to, hint, func(ctx context.Context, rs ring.ReplicationSet, req *ingester_client.LabelNamesRequest, queryLimiter *limiter.QueryLimiter) ([]interface{}, error) {
return d.ForReplicationSet(ctx, rs, d.cfg.ZoneResultsQuorumMetadata, partialDataEnabled, func(ctx context.Context, client ingester_client.IngesterClient) (interface{}, error) {
resp, err := client.LabelNames(ctx, req)
if err != nil {
return nil, err
}
if err := queryLimiter.AddDataBytes(resp.Size()); err != nil {
return nil, validation.LimitError(err.Error())
}

return resp.LabelNames, nil
})
}, matchers...)
Expand Down Expand Up @@ -1385,15 +1400,15 @@ func (d *Distributor) MetricsForLabelMatchersStream(ctx context.Context, from, t
defer stream.CloseSend() //nolint:errcheck
for {
resp, err := stream.Recv()
if err := queryLimiter.AddDataBytes(resp.Size()); err != nil {
return nil, validation.LimitError(err.Error())
}

if err == io.EOF {
break
} else if err != nil {
return nil, err
}
if err := queryLimiter.AddDataBytes(resp.Size()); err != nil {
return nil, validation.LimitError(err.Error())
}

s := make([][]cortexpb.LabelAdapter, 0, len(resp.Metric))
for _, metric := range resp.Metric {
m := cortexpb.FromLabelAdaptersToMetricWithCopy(metric.Labels)
Expand Down
8 changes: 8 additions & 0 deletions pkg/querier/blocks_store_queryable.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -854,6 +854,7 @@ func (q *blocksStoreQuerier) fetchLabelNamesFromStore(
spanLog = spanlogger.FromContext(ctx)
merrMtx = sync.Mutex{}
merr = multierror.MultiError{}
queryLimiter = limiter.QueryLimiterFromContextWithFallback(ctx)
)

// Concurrently fetch series from all clients.
Expand Down Expand Up @@ -894,6 +895,9 @@ func (q *blocksStoreQuerier) fetchLabelNamesFromStore(
}
return errors.Wrapf(err, "failed to fetch label names from %s", c.RemoteAddress())
}
if dataBytesLimitErr := queryLimiter.AddDataBytes(namesResp.Size()); dataBytesLimitErr != nil {
return validation.LimitError(dataBytesLimitErr.Error())
}

myQueriedBlocks := []ulid.ULID(nil)
if namesResp.Hints != nil {
Expand Down Expand Up @@ -957,6 +961,7 @@ func (q *blocksStoreQuerier) fetchLabelValuesFromStore(
spanLog = spanlogger.FromContext(ctx)
merrMtx = sync.Mutex{}
merr = multierror.MultiError{}
queryLimiter = limiter.QueryLimiterFromContextWithFallback(ctx)
)

// Concurrently fetch series from all clients.
Expand Down Expand Up @@ -997,6 +1002,9 @@ func (q *blocksStoreQuerier) fetchLabelValuesFromStore(
}
return errors.Wrapf(err, "failed to fetch label values from %s", c.RemoteAddress())
}
if dataBytesLimitErr := queryLimiter.AddDataBytes(valuesResp.Size()); dataBytesLimitErr != nil {
return validation.LimitError(dataBytesLimitErr.Error())
}

myQueriedBlocks := []ulid.ULID(nil)
if valuesResp.Hints != nil {
Expand Down
Loading