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

Add max-chunks-bytes-per-query limiter #4216

Merged
merged 10 commits into from
May 27, 2021
Merged

Add max-chunks-bytes-per-query limiter #4216

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
8e544d8
Add max-chunks-bytes-per-query limiter
May 24, 2021
e7fd8d6
Fix for distributor test
May 24, 2021
8011feb
Fix spacing and add pr number to changelog
May 25, 2021
3120569
Add unit test for ingester and address code review comments
May 25, 2021
99c9fbc
Fix chunk bytes unit test and code review comments
May 26, 2021
d823cf8
Fix linter error
May 26, 2021
df08789
Move context to a global to per test for distributor tests. Address o…
May 27, 2021
b89016c
Update config docs
May 27, 2021
0da69c8
Change username to userId
May 27, 2021
7956a53
Fix linter error
May 27, 2021
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
2 changes: 1 addition & 1 deletion CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
- `-alertmanager.receivers-firewall.block.private-addresses` renamed to `-alertmanager.receivers-firewall-block-private-addresses`
* [CHANGE] Change default value of `-server.grpc.keepalive.min-time-between-pings` to `10s` and `-server.grpc.keepalive.ping-without-stream-allowed` to `true`. #4168
* [FEATURE] Querier: Added new `-querier.max-fetched-series-per-query` flag. When Cortex is running with blocks storage, the max series per query limit is enforced in the querier and applies to unique series received from ingesters and store-gateway (long-term storage). #4179
* [FEATURE] Querier: Added new `-querier.max-fetched-chunk-bytes-per-query` flag. When Cortex is running with blocks storage, the max chunk bytes limit is enforced in the querier and limits the size of all aggregated chunks returned from ingesters and blocks storage as bytes for a query. #4216
* [FEATURE] Querier/Ruler: Added new `-querier.max-fetched-chunk-bytes-per-query` flag. When Cortex is running with blocks storage, the max chunk bytes limit is enforced in the querier and ruler and limits the size of all aggregated chunks returned from ingesters and storage as bytes for a query. #4216
* [FEATURE] Alertmanager: Added rate-limits to notifiers. Rate limits used by all integrations can be configured using `-alertmanager.notification-rate-limit`, while per-integration rate limits can be specified via `-alertmanager.notification-rate-limit-per-integration` parameter. Both shared and per-integration limits can be overwritten using overrides mechanism. These limits are applied on individual (per-tenant) alertmanagers. Rate-limited notifications are failed notifications. It is possible to monitor rate-limited notifications via new `cortex_alertmanager_notification_rate_limited_total` metric. #4135 #4163
* [ENHANCEMENT] Alertmanager: introduced new metrics to monitor operation when using `-alertmanager.sharding-enabled`: #4149
* `cortex_alertmanager_state_fetch_replica_state_total`
Expand Down
46 changes: 23 additions & 23 deletions pkg/distributor/distributor_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,6 @@ import (
var (
errFail = fmt.Errorf("Fail")
emptyResponse = &cortexpb.WriteResponse{}
ctx = user.InjectOrgID(context.Background(), "user")
)

func TestConfig_Validate(t *testing.T) {
Expand Down Expand Up @@ -110,6 +109,7 @@ func TestDistributor_Push(t *testing.T) {
lastSeenTimestamp := "cortex_distributor_latest_seen_sample_timestamp_seconds"
distributorAppend := "cortex_distributor_ingester_appends_total"
distributorAppendFailure := "cortex_distributor_ingester_append_failures_total"
ctx := user.InjectOrgID(context.Background(), "user")

type samplesIn struct {
num int
Expand Down Expand Up @@ -380,6 +380,7 @@ func TestDistributor_PushIngestionRateLimiter(t *testing.T) {
expectedError error
}

ctx := user.InjectOrgID(context.Background(), "user")
tests := map[string]struct {
distributors int
ingestionRateStrategy string
Expand Down Expand Up @@ -469,12 +470,14 @@ func TestDistributor_PushIngestionRateLimiter(t *testing.T) {
}

func TestDistributor_PushInstanceLimits(t *testing.T) {

type testPush struct {
samples int
metadata int
expectedError error
}

ctx := user.InjectOrgID(context.Background(), "user")
tests := map[string]struct {
preInflight int
preRateSamples int // initial rate before first push
Expand Down Expand Up @@ -618,7 +621,7 @@ func TestDistributor_PushInstanceLimits(t *testing.T) {
}

func TestDistributor_PushHAInstances(t *testing.T) {
ctx = user.InjectOrgID(context.Background(), "user")
ctx := user.InjectOrgID(context.Background(), "user")

for i, tc := range []struct {
enableTracker bool
Expand Down Expand Up @@ -720,6 +723,7 @@ func TestDistributor_PushHAInstances(t *testing.T) {
func TestDistributor_PushQuery(t *testing.T) {
const shuffleShardSize = 5

ctx := user.InjectOrgID(context.Background(), "user")
nameMatcher := mustEqualMatcher(model.MetricNameLabel, "foo")
barMatcher := mustEqualMatcher("bar", "baz")

Expand Down Expand Up @@ -895,6 +899,7 @@ func TestDistributor_PushQuery(t *testing.T) {
func TestDistributor_QueryStream_ShouldReturnErrorIfMaxChunksPerQueryLimitIsReached(t *testing.T) {
const maxChunksLimit = 30 // Chunks are duplicated due to replication factor.

ctx := user.InjectOrgID(context.Background(), "user")
limits := &validation.Limits{}
flagext.DefaultValues(limits)
limits.MaxChunksPerQuery = maxChunksLimit
Expand Down Expand Up @@ -949,13 +954,10 @@ func TestDistributor_QueryStream_ShouldReturnErrorIfMaxChunksPerQueryLimitIsReac
func TestDistributor_QueryStream_ShouldReturnErrorIfMaxSeriesPerQueryLimitIsReached(t *testing.T) {
const maxSeriesLimit = 10

ctx := user.InjectOrgID(context.Background(), "user")
limits := &validation.Limits{}
flagext.DefaultValues(limits)
ctx = limiter.AddQueryLimiterToContext(ctx, limiter.NewQueryLimiter(maxSeriesLimit, 0))
t.Cleanup(func() {
// Reset the limiter for future tests.
ctx = limiter.AddQueryLimiterToContext(ctx, limiter.NewQueryLimiter(0, 0))
})

// Prepare distributors.
ds, _, r, _ := prepare(t, prepConfig{
Expand Down Expand Up @@ -1004,9 +1006,8 @@ func TestDistributor_QueryStream_ShouldReturnErrorIfMaxSeriesPerQueryLimitIsReac

func TestDistributor_QueryStream_ShouldReturnErrorIfMaxChunkBytesPerQueryLimitIsReached(t *testing.T) {
const seriesToAdd = 10
// This is used to track our initial test series to calculate response chunk size.
const initialSeries = 1

ctx := user.InjectOrgID(context.Background(), "user")
limits := &validation.Limits{}
flagext.DefaultValues(limits)

Expand Down Expand Up @@ -1036,17 +1037,13 @@ func TestDistributor_QueryStream_ShouldReturnErrorIfMaxChunkBytesPerQueryLimitIs

// Use the resulting chunks size to calculate the limit as (series to add + our test series) * the response chunk size.
var responseChunkSize = chunkSizeResponse.ChunksSize()
var maxBytesLimit = (seriesToAdd + initialSeries) * responseChunkSize
var maxBytesLimit = (seriesToAdd) * responseChunkSize

// Update the limiter with the calculated limits.
ctx = limiter.AddQueryLimiterToContext(ctx, limiter.NewQueryLimiter(0, maxBytesLimit))
t.Cleanup(func() {
// Reset the limiter for future tests.
ctx = limiter.AddQueryLimiterToContext(ctx, limiter.NewQueryLimiter(0, 0))
})

// Push a number of series below the max chunk bytes limit.
writeReq = makeWriteRequest(0, seriesToAdd, 0)
// Push a number of series below the max chunk bytes limit. Subtract one for the series added above.
writeReq = makeWriteRequest(0, seriesToAdd-1, 0)
writeRes, err = ds[0].Push(ctx, writeReq)
assert.Equal(t, &cortexpb.WriteResponse{}, writeRes)
assert.Nil(t, err)
Expand All @@ -1055,12 +1052,12 @@ func TestDistributor_QueryStream_ShouldReturnErrorIfMaxChunkBytesPerQueryLimitIs
// exceed it), we expect a query running on all series to succeed.
queryRes, err := ds[0].QueryStream(ctx, math.MinInt32, math.MaxInt32, allSeriesMatchers...)
require.NoError(t, err)
assert.Len(t, queryRes.Chunkseries, seriesToAdd+initialSeries)
assert.Len(t, queryRes.Chunkseries, seriesToAdd)

// Push another series to exceed the chunk bytes limit once we'll query back all series.
writeReq = &cortexpb.WriteRequest{}
writeReq.Timeseries = append(writeReq.Timeseries,
makeWriteRequestTimeseries([]cortexpb.LabelAdapter{{Name: model.MetricNameLabel, Value: "another_series"}}, 0, 0),
makeWriteRequestTimeseries([]cortexpb.LabelAdapter{{Name: model.MetricNameLabel, Value: "another_series_1"}}, 0, 0),
)

writeRes, err = ds[0].Push(ctx, writeReq)
Expand All @@ -1075,7 +1072,7 @@ func TestDistributor_QueryStream_ShouldReturnErrorIfMaxChunkBytesPerQueryLimitIs
}

func TestDistributor_Push_LabelRemoval(t *testing.T) {
ctx = user.InjectOrgID(context.Background(), "user")
ctx := user.InjectOrgID(context.Background(), "user")

type testcase struct {
inputSeries labels.Labels
Expand Down Expand Up @@ -1164,6 +1161,7 @@ func TestDistributor_Push_LabelRemoval(t *testing.T) {
}

func TestDistributor_Push_ShouldGuaranteeShardingTokenConsistencyOverTheTime(t *testing.T) {
ctx := user.InjectOrgID(context.Background(), "user")
tests := map[string]struct {
inputSeries labels.Labels
expectedSeries labels.Labels
Expand Down Expand Up @@ -1239,8 +1237,6 @@ func TestDistributor_Push_ShouldGuaranteeShardingTokenConsistencyOverTheTime(t *
limits.DropLabels = []string{"dropped"}
limits.AcceptHASamples = true

ctx = user.InjectOrgID(context.Background(), "user")

for testName, testData := range tests {
t.Run(testName, func(t *testing.T) {
ds, ingesters, r, _ := prepare(t, prepConfig{
Expand Down Expand Up @@ -1276,6 +1272,8 @@ func TestDistributor_Push_LabelNameValidation(t *testing.T) {
{Name: model.MetricNameLabel, Value: "foo"},
{Name: "999.illegal", Value: "baz"},
}
ctx := user.InjectOrgID(context.Background(), "user")

tests := map[string]struct {
inputLabels labels.Labels
skipLabelNameValidationCfg bool
Expand Down Expand Up @@ -1323,7 +1321,7 @@ func TestDistributor_Push_LabelNameValidation(t *testing.T) {
}

func TestDistributor_Push_ExemplarValidation(t *testing.T) {

ctx := user.InjectOrgID(context.Background(), "user")
manyLabels := []string{model.MetricNameLabel, "test"}
for i := 1; i < 31; i++ {
manyLabels = append(manyLabels, fmt.Sprintf("name_%d", i), fmt.Sprintf("value_%d", i))
Expand Down Expand Up @@ -1370,7 +1368,6 @@ func TestDistributor_Push_ExemplarValidation(t *testing.T) {
numDistributors: 1,
shuffleShardSize: 1,
})

_, err := ds[0].Push(ctx, tc.req)
if tc.errMsg != "" {
fromError, _ := status.FromError(err)
Expand All @@ -1386,6 +1383,7 @@ func BenchmarkDistributor_Push(b *testing.B) {
const (
numSeriesPerRequest = 1000
)
ctx := user.InjectOrgID(context.Background(), "user")

tests := map[string]struct {
prepareConfig func(limits *validation.Limits)
Expand Down Expand Up @@ -1577,6 +1575,7 @@ func BenchmarkDistributor_Push(b *testing.B) {

for testName, testData := range tests {
b.Run(testName, func(b *testing.B) {

// Create an in-memory KV store for the ring with 1 ingester registered.
kvStore := consul.NewInMemoryClient(ring.GetCodec())
err := kvStore.CAS(context.Background(), ring.IngesterRingKey,
Expand Down Expand Up @@ -1651,6 +1650,7 @@ func BenchmarkDistributor_Push(b *testing.B) {
}

func TestSlowQueries(t *testing.T) {
ctx := user.InjectOrgID(context.Background(), "user")
nameMatcher := mustEqualMatcher(model.MetricNameLabel, "foo")
nIngesters := 3
for _, shardByAllLabels := range []bool{true, false} {
Expand Down Expand Up @@ -2604,7 +2604,7 @@ func TestSortLabels(t *testing.T) {
}

func TestDistributor_Push_Relabel(t *testing.T) {
ctx = user.InjectOrgID(context.Background(), "user")
ctx := user.InjectOrgID(context.Background(), "user")

type testcase struct {
inputSeries labels.Labels
Expand Down
20 changes: 11 additions & 9 deletions pkg/distributor/ha_tracker_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ import (
"github.com/prometheus/prometheus/pkg/timestamp"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/weaveworks/common/user"

"github.com/cortexproject/cortex/pkg/cortexpb"
"github.com/cortexproject/cortex/pkg/ring"
Expand Down Expand Up @@ -660,7 +661,8 @@ func TestHATracker_MetricsCleanup(t *testing.T) {
func TestCheckReplicaCleanup(t *testing.T) {
replica := "r1"
cluster := "c1"
user := "user"
userName := "user"
Copy link
Contributor

Choose a reason for hiding this comment

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

[nit] userID.

ctx := user.InjectOrgID(context.Background(), userName)

reg := prometheus.NewPedanticRegistry()

Expand All @@ -678,32 +680,32 @@ func TestCheckReplicaCleanup(t *testing.T) {

now := time.Now()

err = c.checkReplica(context.Background(), user, cluster, replica, now)
err = c.checkReplica(context.Background(), userName, cluster, replica, now)
assert.NoError(t, err)
checkReplicaTimestamp(t, time.Second, c, user, cluster, replica, now)
checkReplicaTimestamp(t, time.Second, c, userName, cluster, replica, now)

// Replica is not marked for deletion yet.
checkReplicaDeletionState(t, time.Second, c, user, cluster, true, true, false)
checkReplicaDeletionState(t, time.Second, c, userName, cluster, true, true, false)

// This will mark replica for deletion (with time.Now())
c.cleanupOldReplicas(ctx, now.Add(1*time.Second))

// Verify marking for deletion.
checkReplicaDeletionState(t, time.Second, c, user, cluster, false, true, true)
checkReplicaDeletionState(t, time.Second, c, userName, cluster, false, true, true)

// This will "revive" the replica.
now = time.Now()
err = c.checkReplica(context.Background(), user, cluster, replica, now)
err = c.checkReplica(context.Background(), userName, cluster, replica, now)
assert.NoError(t, err)
checkReplicaTimestamp(t, time.Second, c, user, cluster, replica, now) // This also checks that entry is not marked for deletion.
checkReplicaTimestamp(t, time.Second, c, userName, cluster, replica, now) // This also checks that entry is not marked for deletion.

// This will mark replica for deletion again (with new time.Now())
c.cleanupOldReplicas(ctx, now.Add(1*time.Second))
checkReplicaDeletionState(t, time.Second, c, user, cluster, false, true, true)
checkReplicaDeletionState(t, time.Second, c, userName, cluster, false, true, true)

// Delete entry marked for deletion completely.
c.cleanupOldReplicas(ctx, time.Now().Add(5*time.Second))
checkReplicaDeletionState(t, time.Second, c, user, cluster, false, false, false)
checkReplicaDeletionState(t, time.Second, c, userName, cluster, false, false, false)

require.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(`
# HELP cortex_ha_tracker_replicas_cleanup_marked_for_deletion_total Number of elected replicas marked for deletion.
Expand Down
4 changes: 1 addition & 3 deletions pkg/util/limiter/query_limiter.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ type QueryLimiter struct {
uniqueSeriesMx sync.Mutex
uniqueSeries map[model.Fingerprint]struct{}

chunkBytesCount *atomic.Int64
chunkBytesCount atomic.Int64

maxSeriesPerQuery int
maxChunkBytesPerQuery int
Copy link
Contributor Author

Choose a reason for hiding this comment

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

This limits us to 2GB (2^31 -1 bytes) per query, is it worth making this an unsigned int which is about 4GB (2^32 bytes) per query or a 64 bit number?

Copy link
Contributor

Choose a reason for hiding this comment

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

int64 please. 4GB is not that much. We may have use cases setting higher limits.

Copy link
Contributor

Choose a reason for hiding this comment

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

On 64-bit systems, int is 64-bit, so this is fine. Note that Cortex officially doesn't support 32-bit systems.

Copy link
Contributor

Choose a reason for hiding this comment

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

I would be explicit like we do everywhere else.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Should we also pass in an int64 at the config/limit.go level? Or is leaving NewQueryLimiter(int, int) and casting the maxChunkBytes value to an int64 ok?

Copy link
Contributor

Choose a reason for hiding this comment

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

I would be explicit like we do everywhere else.

I don't think we're explicit "everywhere else". I think it would make sense to use int here simply because we cannot fit more than max of int into memory anyway (applies for both 32-bit and 64-bit platforms).

Copy link
Contributor

Choose a reason for hiding this comment

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

To your question Tyler, if you go with int64 route, you will need to "extend" that everywhere to avoid losing precision somewhere (ie. in NewQueryLimiter too)

Copy link
Contributor

Choose a reason for hiding this comment

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

Ok. Let's not block on this and keep int.

Expand All @@ -38,8 +38,6 @@ func NewQueryLimiter(maxSeriesPerQuery, maxChunkBytesPerQuery int) *QueryLimiter
uniqueSeriesMx: sync.Mutex{},
uniqueSeries: map[model.Fingerprint]struct{}{},

chunkBytesCount: atomic.NewInt64(0),

maxSeriesPerQuery: maxSeriesPerQuery,
maxChunkBytesPerQuery: maxChunkBytesPerQuery,
}
Expand Down
2 changes: 1 addition & 1 deletion pkg/util/validation/limits.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,7 @@ func (l *Limits) RegisterFlags(f *flag.FlagSet) {
f.IntVar(&l.MaxChunksPerQueryFromStore, "store.query-chunk-limit", 2e6, "Deprecated. Use -querier.max-fetched-chunks-per-query CLI flag and its respective YAML config option instead. Maximum number of chunks that can be fetched in a single query. This limit is enforced when fetching chunks from the long-term storage only. When running the Cortex chunks storage, this limit is enforced in the querier and ruler, while when running the Cortex blocks storage this limit is enforced in the querier, ruler and store-gateway. 0 to disable.")
f.IntVar(&l.MaxChunksPerQuery, "querier.max-fetched-chunks-per-query", 0, "Maximum number of chunks that can be fetched in a single query from ingesters and long-term storage: the total number of actual fetched chunks could be 2x the limit, being independently applied when querying ingesters and long-term storage. This limit is enforced in the ingester (if chunks streaming is enabled), querier, ruler and store-gateway. Takes precedence over the deprecated -store.query-chunk-limit. 0 to disable.")
f.IntVar(&l.MaxFetchedSeriesPerQuery, "querier.max-fetched-series-per-query", 0, "The maximum number of unique series for which a query can fetch samples from each ingesters and blocks storage. This limit is enforced in the querier only when running Cortex with blocks storage. 0 to disable")
f.IntVar(&l.MaxFetchedChunkBytesPerQuery, "querier.max-fetched-chunk-bytes-per-query", 0, "The maximum size of all chunks in bytes for which a query can fetch from each ingester and blocks storage. This limit is enforced in the querier only when running Cortex with blocks storage. 0 to disable.")
f.IntVar(&l.MaxFetchedChunkBytesPerQuery, "querier.max-fetched-chunk-bytes-per-query", 0, "The maximum size of all chunks in bytes that a query can fetch from each ingester and storage. This limit is enforced in the querier and ruler only when running Cortex with blocks storage. 0 to disable.")
f.Var(&l.MaxQueryLength, "store.max-query-length", "Limit the query time range (end - start time). This limit is enforced in the query-frontend (on the received query), in the querier (on the query possibly split by the query-frontend) and in the chunks storage. 0 to disable.")
f.Var(&l.MaxQueryLookback, "querier.max-query-lookback", "Limit how long back data (series and metadata) can be queried, up until <lookback> duration ago. This limit is enforced in the query-frontend, querier and ruler. If the requested time range is outside the allowed range, the request will not fail but will be manipulated to only query data within the allowed time range. 0 to disable.")
f.IntVar(&l.MaxQueryParallelism, "querier.max-query-parallelism", 14, "Maximum number of split queries will be scheduled in parallel by the frontend.")
Expand Down