Skip to content

Commit e85a331

Browse files
justinjung04justinjung04
authored
Query priority (#5605)
* Separate user queue channel into two - normalQueue and highPriorityQueue Signed-off-by: Justin Jung <jungjust@amazon.com> * Add ReservedHighPriorityQueriers Signed-off-by: Justin Jung <jungjust@amazon.com> * Change default priority for all requests to low Signed-off-by: Justin Jung <jungjust@amazon.com> * Update config description Signed-off-by: Justin Jung <jungjust@amazon.com> * Lint Signed-off-by: Justin Jung <jungjust@amazon.com> * Fix test Signed-off-by: Justin Jung <jungjust@amazon.com> * More test fix Signed-off-by: Justin Jung <jungjust@amazon.com> * Add HighPriorityQueries confing + placeholder for IsHighPriorityQuery func Signed-off-by: Justin Jung <jungjust@amazon.com> * Nit Signed-off-by: Justin Jung <jungjust@amazon.com> * Lint Signed-off-by: Justin Jung <jungjust@amazon.com> * Implement IsPriorityQuery Signed-off-by: Justin Jung <jungjust@amazon.com> * Add changelog Signed-off-by: Justin Jung <jungjust@amazon.com> * Pass timestamp as param Signed-off-by: Justin Jung <jungjust@amazon.com> * Parse form so that range query parameters are passed to the roundtrip Signed-off-by: Justin Jung <jungjust@amazon.com> * Address comments Signed-off-by: Justin Jung <jungjust@amazon.com> * Add CompiledRegex to HighPriorityQuery Signed-off-by: Justin Jung <jungjust@amazon.com> * Introduce numbered priority + change config structure Signed-off-by: Justin Jung <jungjust@amazon.com> * Lint Signed-off-by: Justin Jung <jungjust@amazon.com> * Updated docs Signed-off-by: Justin Jung <jungjust@amazon.com> * Skip regex compile if it is match all Signed-off-by: Justin Jung <jungjust@amazon.com> * Updated GetPriority to handle new config structure Signed-off-by: Justin Jung <jungjust@amazon.com> * Add priority queue to user queue Signed-off-by: Justin Jung <jungjust@amazon.com> * Nits + priority queue test added Signed-off-by: Justin Jung <jungjust@amazon.com> * Create user request queue test Signed-off-by: Justin Jung <jungjust@amazon.com> * Add reserved querier logic Signed-off-by: Justin Jung <jungjust@amazon.com> * Fix tests Signed-off-by: Justin Jung <jungjust@amazon.com> * Add priority label to request and queue metrics Signed-off-by: Justin Jung <jungjust@amazon.com> * Fix tests Signed-off-by: Justin Jung <jungjust@amazon.com> * Fix tests Signed-off-by: Justin Jung <jungjust@amazon.com> * Refactor Signed-off-by: Justin Jung <jungjust@amazon.com> * Add more tests Signed-off-by: Justin Jung <jungjust@amazon.com> * Lint Signed-off-by: Justin Jung <jungjust@amazon.com> * Update doc Signed-off-by: Justin Jung <jungjust@amazon.com> * Improve time comparison when assigning priority Signed-off-by: Justin Jung <jungjust@amazon.com> * Make reserved querier to match exact priority + change query length gauge vector in user request queue Signed-off-by: Justin Jung <jungjust@amazon.com> * Bug fix Signed-off-by: Justin Jung <jungjust@amazon.com> * Add comments Signed-off-by: Justin Jung <jungjust@amazon.com> * Address comments Signed-off-by: Justin Jung <jungjust@amazon.com> * Make reserved querier to handle priorities higher or equal Signed-off-by: Justin Jung <jungjust@amazon.com> * Add benchmark tests Signed-off-by: Justin Jung <jungjust@amazon.com> * Update regex to be compiled upon unmarshal Signed-off-by: Justin Jung <jungjust@amazon.com> * Make start and end time check to be skipped if not specified Signed-off-by: Justin Jung <jungjust@amazon.com> * Assign priority before splitting the query Signed-off-by: Justin Jung <jungjust@amazon.com> * Attempt to fix tests Signed-off-by: Justin Jung <jungjust@amazon.com> * Address comments Signed-off-by: Justin Jung <jungjust@amazon.com> * Minor improvements Signed-off-by: Justin Jung <jungjust@amazon.com> * Rename query start end time Signed-off-by: Justin Jung <jungjust@amazon.com> * Improve tests Signed-off-by: Justin Jung <jungjust@amazon.com> * Nit Signed-off-by: Justin Jung <jungjust@amazon.com> --------- Signed-off-by: Justin Jung <jungjust@amazon.com>
1 parent ba730c9 commit e85a331

35 files changed

+1683
-168
lines changed

‎CHANGELOG.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
* [CHANGE] Store Gateway: Add a new fastcache based inmemory index cache. #5619
66
* [CHANGE] Index Cache: Multi level cache backfilling operation becomes async. Added `-blocks-storage.bucket-store.index-cache.multilevel.max-async-concurrency` and `-blocks-storage.bucket-store.index-cache.multilevel.max-async-buffer-size` configs and metric `cortex_store_multilevel_index_cache_backfill_dropped_items_total` for number of dropped items. #5661
77
* [FEATURE] Ingester: Add per-tenant new metric `cortex_ingester_tsdb_data_replay_duration_seconds`. #5477
8+
* [FEATURE] Query Frontend/Scheduler: Add query priority support. #5605
89
* [ENHANCEMENT] Store Gateway: Added `-store-gateway.enabled-tenants` and `-store-gateway.disabled-tenants` to explicitly enable or disable store-gateway for specific tenants. #5638
910
* [ENHANCEMENT] Compactor: Add new compactor metric `cortex_compactor_start_duration_seconds`. #5683
1011
* [ENHANCEMENT] Upgraded Docker base images to `alpine:3.18`. #5684

‎docs/configuration/config-file-reference.md

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3041,6 +3041,18 @@ The `limits_config` configures default and per-tenant limits imposed by Cortex s
30413041
# CLI flag: -frontend.max-outstanding-requests-per-tenant
30423042
[max_outstanding_requests_per_tenant: <int> | default = 100]
30433043

3044+
# Configuration for query priority.
3045+
query_priority:
3046+
# Whether queries are assigned with priorities.
3047+
[enabled: <boolean> | default = false]
3048+
3049+
# Priority assigned to all queries by default. Must be a unique value. Use
3050+
# this as a baseline to make certain queries higher/lower priority.
3051+
[default_priority: <int> | default = 0]
3052+
3053+
# List of priority definitions.
3054+
[priorities: <list of PriorityDef> | default = []]
3055+
30443056
# Duration to delay the evaluation of rules to ensure the underlying metrics
30453057
# have been pushed to Cortex.
30463058
# CLI flag: -ruler.evaluation-delay-duration
@@ -5032,6 +5044,37 @@ otel:
50325044
[tls_insecure_skip_verify: <boolean> | default = false]
50335045
```
50345046
5047+
### `PriorityDef`
5048+
5049+
```yaml
5050+
# Priority level. Must be a unique value.
5051+
[priority: <int> | default = 0]
5052+
5053+
# Number of reserved queriers to handle priorities higher or equal to the
5054+
# priority level. Value between 0 and 1 will be used as a percentage.
5055+
[reserved_queriers: <float> | default = 0]
5056+
5057+
# List of query attributes to assign the priority.
5058+
[query_attributes: <list of QueryAttribute> | default = []]
5059+
```
5060+
5061+
### `QueryAttribute`
5062+
5063+
```yaml
5064+
# Regex that the query string should match. If not set, it won't be checked.
5065+
[regex: <string> | default = ""]
5066+
5067+
# Time window that the query should be within. If not set, it won't be checked.
5068+
time_window:
5069+
# Start of the time window that the query should be within. If set to 0, it
5070+
# won't be checked.
5071+
[start: <int> | default = 0]
5072+
5073+
# End of the time window that the query should be within. If set to 0, it
5074+
# won't be checked.
5075+
[end: <int> | default = 0]
5076+
```
5077+
50355078
### `DisabledRuleGroup`
50365079

50375080
```yaml

‎pkg/cortex/modules.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -487,6 +487,7 @@ func (t *Cortex) initQueryFrontendTripperware() (serv services.Service, err erro
487487
queryAnalyzer,
488488
t.Cfg.Querier.DefaultEvaluationInterval,
489489
t.Cfg.Querier.MaxSubQuerySteps,
490+
t.Cfg.Querier.LookbackDelta,
490491
)
491492

492493
return services.NewIdleService(nil, func(_ error) error {

‎pkg/frontend/config.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,7 @@ func InitFrontend(cfg CombinedFrontendConfig, limits v1.Limits, grpcListenPort i
5959
cfg.FrontendV2.Port = grpcListenPort
6060
}
6161

62-
fr, err := v2.NewFrontend(cfg.FrontendV2, log, reg, retry)
62+
fr, err := v2.NewFrontend(cfg.FrontendV2, limits, log, reg, retry)
6363
return transport.AdaptGrpcRoundTripperToHTTPRoundTripper(fr), nil, fr, err
6464

6565
default:

‎pkg/frontend/transport/handler.go

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -343,6 +343,9 @@ func (f *Handler) reportQueryStats(r *http.Request, userID string, queryString u
343343
if ua := r.Header.Get("User-Agent"); len(ua) > 0 {
344344
logMessage = append(logMessage, "user_agent", ua)
345345
}
346+
if queryPriority := r.Header.Get(util.QueryPriorityHeaderKey); len(queryPriority) > 0 {
347+
logMessage = append(logMessage, "priority", queryPriority)
348+
}
346349

347350
if error != nil {
348351
s, ok := status.FromError(error)

‎pkg/frontend/transport/handler_test.go

Lines changed: 55 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ import (
2121
"github.com/weaveworks/common/user"
2222

2323
querier_stats "github.com/cortexproject/cortex/pkg/querier/stats"
24+
"github.com/cortexproject/cortex/pkg/util"
2425
)
2526

2627
type roundTripperFunc func(*http.Request) (*http.Response, error)
@@ -301,34 +302,63 @@ func TestReportQueryStatsFormat(t *testing.T) {
301302
outputBuf := bytes.NewBuffer(nil)
302303
logger := log.NewSyncLogger(log.NewLogfmtLogger(outputBuf))
303304
handler := NewHandler(HandlerConfig{QueryStatsEnabled: true}, http.DefaultTransport, logger, nil)
304-
305305
userID := "fake"
306-
queryString := url.Values(map[string][]string{"query": {"up"}})
307-
req, err := http.NewRequest(http.MethodGet, "http://localhost:8080/prometheus/api/v1/query", nil)
308-
require.NoError(t, err)
309-
req.Header = http.Header{
310-
"User-Agent": []string{"Grafana"},
311-
}
312-
resp := &http.Response{
313-
ContentLength: 1000,
306+
req, _ := http.NewRequest(http.MethodGet, "http://localhost:8080/prometheus/api/v1/query", nil)
307+
resp := &http.Response{ContentLength: 1000}
308+
responseTime := time.Second
309+
statusCode := http.StatusOK
310+
311+
type testCase struct {
312+
queryString url.Values
313+
queryStats *querier_stats.QueryStats
314+
header http.Header
315+
responseErr error
316+
expectedLog string
314317
}
315-
stats := &querier_stats.QueryStats{
316-
Stats: querier_stats.Stats{
317-
WallTime: 3 * time.Second,
318-
FetchedSeriesCount: 100,
319-
FetchedChunksCount: 200,
320-
FetchedSamplesCount: 300,
321-
FetchedChunkBytes: 1024,
322-
FetchedDataBytes: 2048,
318+
319+
tests := map[string]testCase{
320+
"should not include query and header details if empty": {
321+
expectedLog: `level=info msg="query stats" component=query-frontend method=GET path=/prometheus/api/v1/query response_time=1s query_wall_time_seconds=0 fetched_series_count=0 fetched_chunks_count=0 fetched_samples_count=0 fetched_chunks_bytes=0 fetched_data_bytes=0 status_code=200 response_size=1000`,
322+
},
323+
"should include query length and string at the end": {
324+
queryString: url.Values(map[string][]string{"query": {"up"}}),
325+
expectedLog: `level=info msg="query stats" component=query-frontend method=GET path=/prometheus/api/v1/query response_time=1s query_wall_time_seconds=0 fetched_series_count=0 fetched_chunks_count=0 fetched_samples_count=0 fetched_chunks_bytes=0 fetched_data_bytes=0 status_code=200 response_size=1000 query_length=2 param_query=up`,
326+
},
327+
"should include query stats": {
328+
queryStats: &querier_stats.QueryStats{
329+
Stats: querier_stats.Stats{
330+
WallTime: 3 * time.Second,
331+
FetchedSeriesCount: 100,
332+
FetchedChunksCount: 200,
333+
FetchedSamplesCount: 300,
334+
FetchedChunkBytes: 1024,
335+
FetchedDataBytes: 2048,
336+
},
337+
},
338+
expectedLog: `level=info msg="query stats" component=query-frontend method=GET path=/prometheus/api/v1/query response_time=1s query_wall_time_seconds=3 fetched_series_count=100 fetched_chunks_count=200 fetched_samples_count=300 fetched_chunks_bytes=1024 fetched_data_bytes=2048 status_code=200 response_size=1000`,
339+
},
340+
"should include user agent": {
341+
header: http.Header{"User-Agent": []string{"Grafana"}},
342+
expectedLog: `level=info msg="query stats" component=query-frontend method=GET path=/prometheus/api/v1/query response_time=1s query_wall_time_seconds=0 fetched_series_count=0 fetched_chunks_count=0 fetched_samples_count=0 fetched_chunks_bytes=0 fetched_data_bytes=0 status_code=200 response_size=1000 user_agent=Grafana`,
343+
},
344+
"should include response error": {
345+
responseErr: errors.New("foo_err"),
346+
expectedLog: `level=error msg="query stats" component=query-frontend method=GET path=/prometheus/api/v1/query response_time=1s query_wall_time_seconds=0 fetched_series_count=0 fetched_chunks_count=0 fetched_samples_count=0 fetched_chunks_bytes=0 fetched_data_bytes=0 status_code=200 response_size=1000 error=foo_err`,
347+
},
348+
"should include query priority": {
349+
queryString: url.Values(map[string][]string{"query": {"up"}}),
350+
header: http.Header{util.QueryPriorityHeaderKey: []string{"99"}},
351+
expectedLog: `level=info msg="query stats" component=query-frontend method=GET path=/prometheus/api/v1/query response_time=1s query_wall_time_seconds=0 fetched_series_count=0 fetched_chunks_count=0 fetched_samples_count=0 fetched_chunks_bytes=0 fetched_data_bytes=0 status_code=200 response_size=1000 query_length=2 priority=99 param_query=up`,
323352
},
324353
}
325-
responseErr := errors.New("foo_err")
326-
handler.reportQueryStats(req, userID, queryString, time.Second, stats, responseErr, http.StatusOK, resp)
327354

328-
data, err := io.ReadAll(outputBuf)
329-
require.NoError(t, err)
330-
331-
expectedLog := `level=error msg="query stats" component=query-frontend method=GET path=/prometheus/api/v1/query response_time=1s query_wall_time_seconds=3 fetched_series_count=100 fetched_chunks_count=200 fetched_samples_count=300 fetched_chunks_bytes=1024 fetched_data_bytes=2048 status_code=200 response_size=1000 query_length=2 user_agent=Grafana error=foo_err param_query=up
332-
`
333-
require.Equal(t, expectedLog, string(data))
355+
for testName, testData := range tests {
356+
t.Run(testName, func(t *testing.T) {
357+
req.Header = testData.header
358+
handler.reportQueryStats(req, userID, testData.queryString, responseTime, testData.queryStats, testData.responseErr, statusCode, resp)
359+
data, err := io.ReadAll(outputBuf)
360+
require.NoError(t, err)
361+
require.Equal(t, testData.expectedLog+"\n", string(data))
362+
})
363+
}
334364
}

‎pkg/frontend/v1/frontend.go

Lines changed: 24 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import (
55
"flag"
66
"fmt"
77
"net/http"
8+
"strconv"
89
"time"
910

1011
"github.com/go-kit/log"
@@ -51,14 +52,19 @@ type Limits interface {
5152

5253
// MockLimits implements the Limits interface. Used in tests only.
5354
type MockLimits struct {
54-
Queriers float64
55+
Queriers float64
56+
queryPriority validation.QueryPriority
5557
queue.MockLimits
5658
}
5759

5860
func (l MockLimits) MaxQueriersPerUser(_ string) float64 {
5961
return l.Queriers
6062
}
6163

64+
func (l MockLimits) QueryPriority(_ string) validation.QueryPriority {
65+
return l.queryPriority
66+
}
67+
6268
// Frontend queues HTTP requests, dispatches them to backends, and handles retries
6369
// for requests which failed.
6470
type Frontend struct {
@@ -93,6 +99,15 @@ type request struct {
9399
response chan *httpgrpc.HTTPResponse
94100
}
95101

102+
func (r request) Priority() int64 {
103+
priority, err := strconv.ParseInt(httpgrpcutil.GetHeader(*r.request, util.QueryPriorityHeaderKey), 10, 64)
104+
if err != nil {
105+
return 0
106+
}
107+
108+
return priority
109+
}
110+
96111
// New creates a new frontend. Frontend implements service, and must be started and stopped.
97112
func New(cfg Config, limits Limits, log log.Logger, registerer prometheus.Registerer, retry *transport.Retry) (*Frontend, error) {
98113
f := &Frontend{
@@ -103,11 +118,11 @@ func New(cfg Config, limits Limits, log log.Logger, registerer prometheus.Regist
103118
queueLength: promauto.With(registerer).NewGaugeVec(prometheus.GaugeOpts{
104119
Name: "cortex_query_frontend_queue_length",
105120
Help: "Number of queries in the queue.",
106-
}, []string{"user"}),
121+
}, []string{"user", "priority", "type"}),
107122
discardedRequests: promauto.With(registerer).NewCounterVec(prometheus.CounterOpts{
108123
Name: "cortex_query_frontend_discarded_requests_total",
109124
Help: "Total number of query requests discarded.",
110-
}, []string{"user"}),
125+
}, []string{"user", "priority"}),
111126
queueDuration: promauto.With(registerer).NewHistogram(prometheus.HistogramOpts{
112127
Name: "cortex_query_frontend_queue_duration_seconds",
113128
Help: "Time spend by requests queued.",
@@ -160,8 +175,12 @@ func (f *Frontend) stopping(_ error) error {
160175
}
161176

162177
func (f *Frontend) cleanupInactiveUserMetrics(user string) {
163-
f.queueLength.DeleteLabelValues(user)
164-
f.discardedRequests.DeleteLabelValues(user)
178+
f.queueLength.DeletePartialMatch(prometheus.Labels{
179+
"user": user,
180+
})
181+
f.discardedRequests.DeletePartialMatch(prometheus.Labels{
182+
"user": user,
183+
})
165184
}
166185

167186
// RoundTripGRPC round trips a proto (instead of a HTTP request).

‎pkg/frontend/v1/frontend_test.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -211,7 +211,7 @@ func TestFrontendMetricsCleanup(t *testing.T) {
211211
require.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(`
212212
# HELP cortex_query_frontend_queue_length Number of queries in the queue.
213213
# TYPE cortex_query_frontend_queue_length gauge
214-
cortex_query_frontend_queue_length{user="1"} 0
214+
cortex_query_frontend_queue_length{priority="0",type="fifo",user="1"} 0
215215
`), "cortex_query_frontend_queue_length"))
216216

217217
fr.cleanupInactiveUserMetrics("1")

‎pkg/frontend/v2/frontend.go

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ import (
2121
"github.com/cortexproject/cortex/pkg/frontend/transport"
2222
"github.com/cortexproject/cortex/pkg/frontend/v2/frontendv2pb"
2323
"github.com/cortexproject/cortex/pkg/querier/stats"
24+
"github.com/cortexproject/cortex/pkg/scheduler"
2425
"github.com/cortexproject/cortex/pkg/tenant"
2526
"github.com/cortexproject/cortex/pkg/util/flagext"
2627
"github.com/cortexproject/cortex/pkg/util/grpcclient"
@@ -64,10 +65,10 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
6465
type Frontend struct {
6566
services.Service
6667

67-
cfg Config
68-
log log.Logger
69-
70-
retry *transport.Retry
68+
cfg Config
69+
log log.Logger
70+
limits scheduler.Limits
71+
retry *transport.Retry
7172

7273
lastQueryID atomic.Uint64
7374

@@ -112,7 +113,7 @@ type enqueueResult struct {
112113
}
113114

114115
// NewFrontend creates a new frontend.
115-
func NewFrontend(cfg Config, log log.Logger, reg prometheus.Registerer, retry *transport.Retry) (*Frontend, error) {
116+
func NewFrontend(cfg Config, limits scheduler.Limits, log log.Logger, reg prometheus.Registerer, retry *transport.Retry) (*Frontend, error) {
116117
requestsCh := make(chan *frontendRequest)
117118

118119
schedulerWorkers, err := newFrontendSchedulerWorkers(cfg, fmt.Sprintf("%s:%d", cfg.Addr, cfg.Port), requestsCh, log)
@@ -122,6 +123,7 @@ func NewFrontend(cfg Config, log log.Logger, reg prometheus.Registerer, retry *t
122123

123124
f := &Frontend{
124125
cfg: cfg,
126+
limits: limits,
125127
log: log,
126128
requestsCh: requestsCh,
127129
schedulerWorkers: schedulerWorkers,

‎pkg/frontend/v2/frontend_test.go

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ import (
1919
"github.com/cortexproject/cortex/pkg/frontend/transport"
2020
"github.com/cortexproject/cortex/pkg/frontend/v2/frontendv2pb"
2121
"github.com/cortexproject/cortex/pkg/querier/stats"
22+
"github.com/cortexproject/cortex/pkg/scheduler/queue"
2223
"github.com/cortexproject/cortex/pkg/scheduler/schedulerpb"
2324
"github.com/cortexproject/cortex/pkg/util/flagext"
2425
"github.com/cortexproject/cortex/pkg/util/services"
@@ -48,7 +49,7 @@ func setupFrontend(t *testing.T, schedulerReplyFunc func(f *Frontend, msg *sched
4849

4950
//logger := log.NewLogfmtLogger(os.Stdout)
5051
logger := log.NewNopLogger()
51-
f, err := NewFrontend(cfg, logger, nil, transport.NewRetry(maxRetries, nil))
52+
f, err := NewFrontend(cfg, queue.MockLimits{}, logger, nil, transport.NewRetry(maxRetries, nil))
5253
require.NoError(t, err)
5354

5455
frontendv2pb.RegisterFrontendForQuerierServer(server, f)

‎pkg/querier/tripperware/instantquery/instant_query.go

Lines changed: 1 addition & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -8,14 +8,12 @@ import (
88
"net/http"
99
"net/url"
1010
"sort"
11-
"strconv"
1211
"strings"
1312
"time"
1413

1514
jsoniter "github.com/json-iterator/go"
1615
"github.com/opentracing/opentracing-go"
1716
otlog "github.com/opentracing/opentracing-go/log"
18-
"github.com/pkg/errors"
1917
"github.com/prometheus/common/model"
2018
"github.com/prometheus/prometheus/model/labels"
2119
"github.com/prometheus/prometheus/model/timestamp"
@@ -132,7 +130,7 @@ func (resp *PrometheusInstantQueryResponse) HTTPHeaders() map[string][]string {
132130
func (c instantQueryCodec) DecodeRequest(_ context.Context, r *http.Request, forwardHeaders []string) (tripperware.Request, error) {
133131
result := PrometheusRequest{Headers: map[string][]string{}}
134132
var err error
135-
result.Time, err = parseTimeParam(r, "time", c.now().Unix())
133+
result.Time, err = util.ParseTimeParam(r, "time", c.now().Unix())
136134
if err != nil {
137135
return nil, decorateWithParamName(err, "time")
138136
}
@@ -630,15 +628,3 @@ func (s *PrometheusInstantQueryData) MarshalJSON() ([]byte, error) {
630628
return s.Result.GetRawBytes(), nil
631629
}
632630
}
633-
634-
func parseTimeParam(r *http.Request, paramName string, defaultValue int64) (int64, error) {
635-
val := r.FormValue(paramName)
636-
if val == "" {
637-
val = strconv.FormatInt(defaultValue, 10)
638-
}
639-
result, err := util.ParseTime(val)
640-
if err != nil {
641-
return 0, errors.Wrapf(err, "Invalid time value for '%s'", paramName)
642-
}
643-
return result, nil
644-
}

‎pkg/querier/tripperware/limits.go

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,10 @@
11
package tripperware
22

3-
import "time"
3+
import (
4+
"time"
5+
6+
"github.com/cortexproject/cortex/pkg/util/validation"
7+
)
48

59
// Limits allows us to specify per-tenant runtime limits on the behavior of
610
// the query handling code.
@@ -21,4 +25,7 @@ type Limits interface {
2125

2226
// QueryVerticalShardSize returns the maximum number of queriers that can handle requests for this user.
2327
QueryVerticalShardSize(userID string) int
28+
29+
// QueryPriority returns the query priority config for the tenant, including different priorities and their attributes.
30+
QueryPriority(userID string) validation.QueryPriority
2431
}

0 commit comments

Comments
 (0)