Add multi tenant query federation #3250
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,181 @@ | ||
| // +build requires_docker | ||
|
|
||
| package integration | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "strings" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/prometheus/common/model" | ||
| "github.com/prometheus/prometheus/pkg/labels" | ||
| "github.com/prometheus/prometheus/prompb" | ||
| "github.com/stretchr/testify/assert" | ||
| "github.com/stretchr/testify/require" | ||
|
|
||
| "github.com/cortexproject/cortex/integration/e2e" | ||
| e2ecache "github.com/cortexproject/cortex/integration/e2e/cache" | ||
| e2edb "github.com/cortexproject/cortex/integration/e2e/db" | ||
| "github.com/cortexproject/cortex/integration/e2ecortex" | ||
| ) | ||
|
|
||
| type querierTenantFederationConfig struct { | ||
| querySchedulerEnabled bool | ||
| shuffleShardingEnabled bool | ||
| } | ||
|
|
||
| func TestQuerierTenantFederation(t *testing.T) { | ||
| runQuerierTenantFederationTest(t, querierTenantFederationConfig{}) | ||
| } | ||
|
|
||
| func TestQuerierTenantFederationWithQueryScheduler(t *testing.T) { | ||
| runQuerierTenantFederationTest(t, querierTenantFederationConfig{ | ||
| querySchedulerEnabled: true, | ||
| }) | ||
| } | ||
|
|
||
| func TestQuerierTenantFederationWithShuffleSharding(t *testing.T) { | ||
| runQuerierTenantFederationTest(t, querierTenantFederationConfig{ | ||
| shuffleShardingEnabled: true, | ||
| }) | ||
| } | ||
|
|
||
| func TestQuerierTenantFederationWithQuerySchedulerAndShuffleSharding(t *testing.T) { | ||
| runQuerierTenantFederationTest(t, querierTenantFederationConfig{ | ||
| querySchedulerEnabled: true, | ||
| shuffleShardingEnabled: true, | ||
| }) | ||
| } | ||
|
|
||
| func runQuerierTenantFederationTest(t *testing.T, cfg querierTenantFederationConfig) { | ||
| const numUsers = 10 | ||
|
|
||
| s, err := e2e.NewScenario(networkName) | ||
| require.NoError(t, err) | ||
| defer s.Close() | ||
|
|
||
| memcached := e2ecache.NewMemcached() | ||
| consul := e2edb.NewConsul() | ||
| require.NoError(t, s.StartAndWaitReady(consul, memcached)) | ||
|
|
||
| flags := mergeFlags(BlocksStorageFlags(), map[string]string{ | ||
| "-querier.cache-results": "true", | ||
| "-querier.split-queries-by-interval": "24h", | ||
| "-querier.query-ingesters-within": "12h", // Required by the test on query /series out of ingesters time range | ||
| "-frontend.memcached.addresses": "dns+" + memcached.NetworkEndpoint(e2ecache.MemcachedPort), | ||
| "-tenant-federation.enabled": "true", | ||
|
There was a problem hiding this comment. thought: We should be able to enable this feature for every e2e test and not have any failures. Doing so could provide a good smoke test for the feature. There was a problem hiding this comment. @jtlisi Not entirely sure if you meant to do this locally once or do you want to have that as part of the integration test run by CI? There was a problem hiding this comment. This is still a TODO, will look into this tomorrow There was a problem hiding this comment. No need to commit any changes here. Just smoke testing that the tests pass if you enable it for globally on your local should be enough. There was a problem hiding this comment. I have done that today and it didn't not come up with any unexpected failure (apart from the |
||
| }) | ||
|
|
||
| // Start the query-scheduler if enabled. | ||
| var queryScheduler *e2ecortex.CortexService | ||
| if cfg.querySchedulerEnabled { | ||
| queryScheduler = e2ecortex.NewQueryScheduler("query-scheduler", flags, "") | ||
| require.NoError(t, s.StartAndWaitReady(queryScheduler)) | ||
| flags["-frontend.scheduler-address"] = queryScheduler.NetworkGRPCEndpoint() | ||
| flags["-querier.scheduler-address"] = queryScheduler.NetworkGRPCEndpoint() | ||
| } | ||
|
|
||
| if cfg.shuffleShardingEnabled { | ||
| // Use only single querier for each user. | ||
| flags["-frontend.max-queriers-per-tenant"] = "1" | ||
| } | ||
|
|
||
| minio := e2edb.NewMinio(9000, flags["-blocks-storage.s3.bucket-name"]) | ||
| require.NoError(t, s.StartAndWaitReady(minio)) | ||
|
|
||
| // Start the query-frontend. | ||
| queryFrontend := e2ecortex.NewQueryFrontend("query-frontend", flags, "") | ||
| require.NoError(t, s.Start(queryFrontend)) | ||
|
|
||
| if !cfg.querySchedulerEnabled { | ||
| flags["-querier.frontend-address"] = queryFrontend.NetworkGRPCEndpoint() | ||
| } | ||
|
|
||
| // Start all other services. | ||
| ingester := e2ecortex.NewIngester("ingester", consul.NetworkHTTPEndpoint(), flags, "") | ||
| distributor := e2ecortex.NewDistributor("distributor", consul.NetworkHTTPEndpoint(), flags, "") | ||
| querier := e2ecortex.NewQuerier("querier", consul.NetworkHTTPEndpoint(), flags, "") | ||
|
|
||
| var querier2 *e2ecortex.CortexService | ||
| if cfg.shuffleShardingEnabled { | ||
| querier2 = e2ecortex.NewQuerier("querier-2", consul.NetworkHTTPEndpoint(), flags, "") | ||
| } | ||
|
|
||
| require.NoError(t, s.StartAndWaitReady(querier, ingester, distributor)) | ||
| require.NoError(t, s.WaitReady(queryFrontend)) | ||
| if cfg.shuffleShardingEnabled { | ||
| require.NoError(t, s.StartAndWaitReady(querier2)) | ||
| } | ||
|
|
||
| // Wait until distributor and queriers have updated the ring. | ||
| require.NoError(t, distributor.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total")) | ||
| require.NoError(t, querier.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total")) | ||
| if cfg.shuffleShardingEnabled { | ||
| require.NoError(t, querier2.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total")) | ||
| } | ||
|
|
||
| // Push a series for each user to Cortex. | ||
| now := time.Now() | ||
| expectedVectors := make([]model.Vector, numUsers) | ||
| tenantIDs := make([]string, numUsers) | ||
|
|
||
| for u := 0; u < numUsers; u++ { | ||
| tenantIDs[u] = fmt.Sprintf("user-%d", u) | ||
| c, err := e2ecortex.NewClient(distributor.HTTPEndpoint(), "", "", "", tenantIDs[u]) | ||
| require.NoError(t, err) | ||
|
|
||
| var series []prompb.TimeSeries | ||
| series, expectedVectors[u] = generateSeries("series_1", now) | ||
|
|
||
| res, err := c.Push(series) | ||
| require.NoError(t, err) | ||
| require.Equal(t, 200, res.StatusCode) | ||
| } | ||
|
|
||
| // query all tenants | ||
| c, err := e2ecortex.NewClient(distributor.HTTPEndpoint(), queryFrontend.HTTPEndpoint(), "", "", strings.Join(tenantIDs, "|")) | ||
| require.NoError(t, err) | ||
|
|
||
| result, err := c.Query("series_1", now) | ||
| require.NoError(t, err) | ||
|
|
||
| assert.Equal(t, mergeResults(tenantIDs, expectedVectors), result.(model.Vector)) | ||
|
|
||
| // ensure a push to multiple tenants is failing | ||
| series, _ := generateSeries("series_1", now) | ||
| res, err := c.Push(series) | ||
| require.NoError(t, err) | ||
| require.Equal(t, 500, res.StatusCode) | ||
|
|
||
| // check metric label values for total queries in the query frontend | ||
| require.NoError(t, queryFrontend.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"cortex_query_frontend_queries_total"}, e2e.WithLabelMatchers( | ||
| labels.MustNewMatcher(labels.MatchEqual, "user", strings.Join(tenantIDs, "|")), | ||
| labels.MustNewMatcher(labels.MatchEqual, "op", "query")))) | ||
|
|
||
| // check metric label values for query queue length in either query frontend or query scheduler | ||
| queueComponent := queryFrontend | ||
| queueMetricName := "cortex_query_frontend_queue_length" | ||
| if cfg.querySchedulerEnabled { | ||
| queueComponent = queryScheduler | ||
| queueMetricName = "cortex_query_scheduler_queue_length" | ||
| } | ||
| require.NoError(t, queueComponent.WaitSumMetricsWithOptions(e2e.Equals(0), []string{queueMetricName}, e2e.WithLabelMatchers( | ||
| labels.MustNewMatcher(labels.MatchEqual, "user", strings.Join(tenantIDs, "|"))))) | ||
|
|
||
| // TODO: check cache invalidation on tombstone cache gen increase | ||
| // TODO: check fairness in queryfrontend | ||
simonswine marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| } | ||
|
|
||
| func mergeResults(tenantIDs []string, resultsPerTenant []model.Vector) model.Vector { | ||
| var v model.Vector | ||
| for pos, tenantID := range tenantIDs { | ||
| for _, r := range resultsPerTenant[pos] { | ||
| var s model.Sample = *r | ||
| s.Metric = r.Metric.Clone() | ||
| s.Metric[model.LabelName("__tenant_id__")] = model.LabelValue(tenantID) | ||
| v = append(v, &s) | ||
| } | ||
| } | ||
| return v | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -30,6 +30,7 @@ import ( | |
| "github.com/cortexproject/cortex/pkg/ingester" | ||
| "github.com/cortexproject/cortex/pkg/querier" | ||
| "github.com/cortexproject/cortex/pkg/querier/queryrange" | ||
| "github.com/cortexproject/cortex/pkg/querier/tenantfederation" | ||
| querier_worker "github.com/cortexproject/cortex/pkg/querier/worker" | ||
| "github.com/cortexproject/cortex/pkg/ring" | ||
| "github.com/cortexproject/cortex/pkg/ring/kv/codec" | ||
|
|
@@ -203,7 +204,15 @@ func (t *Cortex) initQueryable() (serv services.Service, err error) { | |
| querierRegisterer := prometheus.WrapRegistererWith(prometheus.Labels{"engine": "querier"}, prometheus.DefaultRegisterer) | ||
|
|
||
| // Create a querier queryable and PromQL engine | ||
| var queryable prom_storage.SampleAndChunkQueryable | ||
| queryable, t.QuerierEngine = querier.New(t.Cfg.Querier, t.Overrides, t.Distributor, t.StoreQueryables, t.TombstonesLoader, querierRegisterer) | ||
|
|
||
| // Enable merge querier if multi tenant query federation is enabled | ||
| if t.Cfg.TenantFederation.Enabled { | ||
| queryable = querier.NewSampleAndChunkQueryable(tenantfederation.NewQueryable(queryable)) | ||
|
There was a problem hiding this comment. May you confirm we don't need to do this for the ruler as well, because the ruler will always run on a single tenant? There was a problem hiding this comment. The proposal lists that under future work , right now I think we should not enable that just yet. |
||
| } | ||
|
|
||
| t.QuerierQueryable = queryable | ||
|
|
||
| // Register the default endpoints that are always enabled for the querier module | ||
| t.API.RegisterQueryable(t.QuerierQueryable, t.Distributor) | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why all? Let's be specific to avoid confusing users.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think it needs to be enabled on all Cortex services, because otherwise e.g. the write path would interpret multiple tenants as a single new tenant. Once it is enabled the write path fails as it only supports a single tenant. I have aligned the CHANGELOG with the flags description.