Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
With the current GRPC APIs, layering Thanos Queriers results in the root querier getting all of the samples and executing the query in memory. As a result, the intermediary Queriers do not do any intensive work and merely transport samples from the Stores to the root Querier. When data is perfectly sharded, users can implement a pattern where the root Querier instructs the intermediary ones to execute the queries from their stores and return back results. The results can then be concatenated by the root querier and returned to the user. In order to support this use case, this commit implements a GRPC API in the Querier which is analogous to the HTTP Query API exposed by Prometheus. Signed-off-by: fpetkovski <filip.petkovsky@gmail.com>
- Loading branch information
1 parent
149e026
commit 48fff63
Showing
12 changed files
with
2,308 additions
and
5 deletions.
There are no files selected for viewing
This file contains 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 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 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 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 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 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,161 @@ | ||
| // Copyright (c) The Thanos Authors. | ||
| // Licensed under the Apache License 2.0. | ||
|
|
||
| package v1 | ||
|
|
||
| import ( | ||
| "context" | ||
| "time" | ||
|
|
||
| "github.com/prometheus/prometheus/promql" | ||
| "github.com/thanos-io/thanos/pkg/api/query/querypb" | ||
| "github.com/thanos-io/thanos/pkg/query" | ||
| "github.com/thanos-io/thanos/pkg/store/labelpb" | ||
| "github.com/thanos-io/thanos/pkg/store/storepb/prompb" | ||
| "google.golang.org/grpc" | ||
| ) | ||
|
|
||
| type GrpcAPI struct { | ||
| now func() time.Time | ||
| queryableCreate query.QueryableCreator | ||
| queryEngine func(int64) *promql.Engine | ||
| defaultMaxResolutionSeconds time.Duration | ||
| } | ||
|
|
||
| func NewGrpcAPI(now func() time.Time, creator query.QueryableCreator, queryEngine func(int64) *promql.Engine, defaultMaxResolutionSeconds time.Duration) *GrpcAPI { | ||
| return &GrpcAPI{ | ||
| now: now, | ||
| queryableCreate: creator, | ||
| queryEngine: queryEngine, | ||
| defaultMaxResolutionSeconds: defaultMaxResolutionSeconds, | ||
| } | ||
| } | ||
|
|
||
| func RegisterQueryServer(queryServer querypb.QueryServer) func(*grpc.Server) { | ||
| return func(s *grpc.Server) { | ||
| querypb.RegisterQueryServer(s, queryServer) | ||
| } | ||
| } | ||
|
|
||
| func (grpcAPI *GrpcAPI) Query(ctx context.Context, request *querypb.QueryRequest) (*querypb.QueryResponse, error) { | ||
| var ts time.Time | ||
| if request.TimeSeconds == 0 { | ||
| ts = grpcAPI.now() | ||
| } else { | ||
| ts = time.Unix(request.TimeSeconds, 0) | ||
| } | ||
|
|
||
| if request.TimeoutSeconds != 0 { | ||
| var cancel context.CancelFunc | ||
| timeout := time.Duration(request.TimeoutSeconds) * time.Second | ||
| ctx, cancel = context.WithTimeout(ctx, timeout) | ||
| defer cancel() | ||
| } | ||
|
|
||
| maxResolution := request.MaxResolutionSeconds | ||
| if request.MaxResolutionSeconds == 0 { | ||
| maxResolution = grpcAPI.defaultMaxResolutionSeconds.Milliseconds() / 1000 | ||
| } | ||
|
|
||
| storeMatchers, err := querypb.StoreMatchersToLabelMatchers(request.StoreMatchers) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| qe := grpcAPI.queryEngine(request.MaxResolutionSeconds) | ||
| queryable := grpcAPI.queryableCreate( | ||
| request.EnableDedup, | ||
| request.ReplicaLabels, | ||
| storeMatchers, | ||
| maxResolution, | ||
| request.EnablePartialResponse, | ||
| request.EnableQueryPushdown, | ||
| false, | ||
| ) | ||
| qry, err := qe.NewInstantQuery(queryable, request.Query, ts) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| result := qry.Exec(ctx) | ||
| switch vector := result.Value.(type) { | ||
| case promql.Scalar: | ||
| return &querypb.QueryResponse{ | ||
| Timeseries: []prompb.TimeSeries{{ | ||
| Samples: []prompb.Sample{{Value: vector.V, Timestamp: vector.T}}, | ||
| }}, | ||
| }, nil | ||
| case promql.Vector: | ||
| response := &querypb.QueryResponse{ | ||
| Timeseries: make([]prompb.TimeSeries, 0, len(vector)), | ||
| } | ||
|
|
||
| for _, sample := range vector { | ||
| response.Timeseries = append(response.Timeseries, prompb.TimeSeries{ | ||
| Labels: labelpb.ZLabelsFromPromLabels(sample.Metric), | ||
| Samples: prompb.SamplesFromPromqlPoints([]promql.Point{sample.Point}), | ||
| }) | ||
| } | ||
|
|
||
| return response, nil | ||
| } | ||
|
|
||
| return nil, nil | ||
| } | ||
|
|
||
| func (grpcAPI *GrpcAPI) QueryRange(ctx context.Context, request *querypb.QueryRangeRequest) (*querypb.QueryRangeResponse, error) { | ||
| if request.TimeoutSeconds != 0 { | ||
| var cancel context.CancelFunc | ||
| ctx, cancel = context.WithTimeout(ctx, time.Duration(request.TimeoutSeconds)) | ||
| defer cancel() | ||
| } | ||
|
|
||
| maxResolution := request.MaxResolutionSeconds | ||
| if request.MaxResolutionSeconds == 0 { | ||
| maxResolution = grpcAPI.defaultMaxResolutionSeconds.Milliseconds() / 1000 | ||
| } | ||
|
|
||
| storeMatchers, err := querypb.StoreMatchersToLabelMatchers(request.StoreMatchers) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| qe := grpcAPI.queryEngine(request.MaxResolutionSeconds) | ||
| queryable := grpcAPI.queryableCreate( | ||
| request.EnableDedup, | ||
| request.ReplicaLabels, | ||
| storeMatchers, | ||
| maxResolution, | ||
| request.EnablePartialResponse, | ||
| request.EnableQueryPushdown, | ||
| false, | ||
| ) | ||
|
|
||
| startTime := time.Unix(request.StartTimeSeconds, 0) | ||
| endTime := time.Unix(request.EndTimeSeconds, 0) | ||
| interval := time.Duration(request.IntervalSeconds) * time.Second | ||
|
|
||
| qry, err := qe.NewRangeQuery(queryable, request.Query, startTime, endTime, interval) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| result := qry.Exec(ctx) | ||
| switch matrix := result.Value.(type) { | ||
| case promql.Matrix: | ||
| response := &querypb.QueryRangeResponse{ | ||
| Timeseries: make([]prompb.TimeSeries, len(matrix)), | ||
| } | ||
|
|
||
| for i, series := range matrix { | ||
| response.Timeseries[i] = prompb.TimeSeries{ | ||
| Labels: labelpb.ZLabelsFromPromLabels(series.Metric), | ||
| Samples: prompb.SamplesFromPromqlPoints(series.Points), | ||
| } | ||
| } | ||
|
|
||
| return response, nil | ||
| } | ||
|
|
||
| return nil, nil | ||
| } |
Oops, something went wrong.