public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Run multiple queries as one query trough backend

Browse Source
This commit is contained in:
Ivana
2021-09-21 18:12:19 +02:00
parent ed938fd006
commit e3f5c4b57a
6 changed files with 156 additions and 187 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/tsdb/prometheus/prometheus.go
+77 -86
View File
@@ -34,26 +34,6 @@ var (
safeRes = 11000
)
type DatasourceInfo struct {
ID int64
HTTPClientOpts sdkhttpclient.Options
URL string
HTTPMethod string
TimeInterval string
}
type QueryModel struct {
Expr string `json:"expr"`
LegendFormat string `json:"legendFormat"`
Interval string `json:"interval"`
IntervalMS int64 `json:"intervalMS"`
StepMode string `json:"stepMode"`
RangeQuery bool `json:"range"`
InstantQuery bool `json:"instant"`
IntervalFactor int64 `json:"intervalFactor"`
UtcOffsetSec int64 `json:"utcOffsetSec"`
}
type Service struct {
httpClientProvider httpclient.Provider
intervalCalculator intervalv2.Calculator
@@ -149,13 +129,7 @@ func (s *Service) QueryData(ctx context.Context, req *backend.QueryDataRequest)
}
for _, query := range queries {
timeRange := apiv1.Range{
Start: query.Start,
End: query.End,
Step: query.Step,
}
plog.Debug("Sending query", "start", timeRange.Start, "end", timeRange.End, "step", timeRange.Step, "query", query.Expr)
plog.Debug("Sending query", "start", query.Start, "end", query.End, "step", query.Step, "query", query.Expr)
span, ctx := opentracing.StartSpanFromContext(ctx, "datasource.prometheus")
span.SetTag("expr", query.Expr)
@@ -163,30 +137,39 @@ func (s *Service) QueryData(ctx context.Context, req *backend.QueryDataRequest)
span.SetTag("stop_unixnano", query.End.UnixNano())
defer span.Finish()
var response model.Value
response := make(map[PrometheusQueryType]model.Value)
switch query.QueryType {
case Range:
response, _, err = client.QueryRange(ctx, query.Expr, timeRange)
if query.RangeQuery {
timeRange := apiv1.Range{
Step: query.Step,
// Align query range to step. It rounds start and end down to a multiple of step.
Start: time.Unix(int64(math.Floor((float64(query.Start.Unix()+query.UtcOffsetSec)/query.Step.Seconds()))*query.Step.Seconds()-float64(query.UtcOffsetSec)), 0),
End: time.Unix(int64(math.Floor((float64(query.End.Unix()+query.UtcOffsetSec)/query.Step.Seconds()))*query.Step.Seconds()-float64(query.UtcOffsetSec)), 0),
}
rangeResponse, _, err := client.QueryRange(ctx, query.Expr, timeRange)
if err != nil {
return &result, fmt.Errorf("query: %s failed with: %v", query.Expr, err)
}
case Instant:
response, _, err = client.Query(ctx, query.Expr, query.End)
if err != nil {
return &result, fmt.Errorf("query: %s failed with: %v", query.Expr, err)
}
default:
return &result, fmt.Errorf("unknown Query type detected %#v", query.QueryType)
response[Range] = rangeResponse
}
frame, err := parseResponse(response, query)
if query.InstantQuery {
instantResponse, _, err := client.Query(ctx, query.Expr, query.End)
if err != nil {
return &result, fmt.Errorf("query: %s failed with: %v", query.Expr, err)
}
response[Instant] = instantResponse
}
frames, err := parseResponse(response, query)
if err != nil {
return &result, err
}
result.Responses[query.RefId] = backend.DataResponse{
Frames: frame,
Frames: frames,
}
}
@@ -304,60 +287,49 @@ func (s *Service) parseQuery(queryContext *backend.QueryDataRequest, dsInfo *Dat
expr = strings.ReplaceAll(expr, "$__range", strconv.FormatInt(rangeS, 10)+"s")
expr = strings.ReplaceAll(expr, "$__rate_interval", intervalv2.FormatDuration(calculateRateInterval(interval, dsInfo.TimeInterval, s.intervalCalculator)))
if model.RangeQuery && model.InstantQuery {
return nil, fmt.Errorf("the provided query is not valid, expected only one of `range` and `instant` to be true")
}
var queryType PrometheusQueryType
var start time.Time
var end time.Time
if model.InstantQuery {
queryType = Instant
start = query.TimeRange.From
end = query.TimeRange.To
} else {
queryType = Range
// Align query range to step. It rounds start and end down to a multiple of step.
start = time.Unix(int64(math.Floor((float64(query.TimeRange.From.Unix()+model.UtcOffsetSec)/interval.Seconds()))*interval.Seconds()-float64(model.UtcOffsetSec)), 0)
end = time.Unix(int64(math.Floor((float64(query.TimeRange.To.Unix()+model.UtcOffsetSec)/interval.Seconds()))*interval.Seconds()-float64(model.UtcOffsetSec)), 0)
rangeQuery := model.RangeQuery
if !model.InstantQuery && !model.RangeQuery {
// In older dashboards, we were not setting range query param and !range && !instant was run as range query
rangeQuery = true
}
qs = append(qs, &PrometheusQuery{
Expr: expr,
Step: interval,
LegendFormat: model.LegendFormat,
Start: start,
End: end,
Start: query.TimeRange.From,
End: query.TimeRange.To,
RefId: query.RefID,
QueryType: queryType,
InstantQuery: model.InstantQuery,
RangeQuery: rangeQuery,
UtcOffsetSec: model.UtcOffsetSec,
})
}
return qs, nil
}
func parseResponse(value model.Value, query *PrometheusQuery) (data.Frames, error) {
func parseResponse(value map[PrometheusQueryType]model.Value, query *PrometheusQuery) (data.Frames, error) {
frames := data.Frames{}
matrix, ok := value.(model.Matrix)
if ok {
matrixFrames := matrixToDataFrames(matrix, query)
frames = append(frames, matrixFrames...)
}
for queryType, value := range value {
matrix, ok := value.(model.Matrix)
if ok {
matrixFrames := matrixToDataFrames(matrix, query, queryType)
frames = append(frames, matrixFrames...)
}
vector, ok := value.(model.Vector)
if ok {
vectorFrames := vectorToDataFrames(vector, query)
frames = append(frames, vectorFrames...)
}
vector, ok := value.(model.Vector)
if ok {
vectorFrames := vectorToDataFrames(vector, query, queryType)
frames = append(frames, vectorFrames...)
}
scalar, ok := value.(*model.Scalar)
if ok {
scalarFrames := scalarToDataFrames(scalar)
frames = append(frames, scalarFrames...)
scalar, ok := value.(*model.Scalar)
if ok {
scalarFrames := scalarToDataFrames(scalar, query, queryType)
frames = append(frames, scalarFrames...)
}
}
return frames, nil
}
@@ -391,7 +363,7 @@ func calculateRateInterval(interval time.Duration, scrapeInterval string, interv
return rateInterval
}
func matrixToDataFrames(matrix model.Matrix, query *PrometheusQuery) data.Frames {
func matrixToDataFrames(matrix model.Matrix, query *PrometheusQuery, queryType PrometheusQueryType) data.Frames {
frames := data.Frames{}
for _, v := range matrix {
@@ -406,26 +378,39 @@ func matrixToDataFrames(matrix model.Matrix, query *PrometheusQuery) data.Frames
values = append(values, float64(k.Value))
}
name := formatLegend(v.Metric, query)
frames = append(frames, data.NewFrame(name,
frame := data.NewFrame(name,
data.NewField("Time", nil, timeVector),
data.NewField("Value", tags, values).SetConfig(&data.FieldConfig{DisplayNameFromDS: name})))
data.NewField("Value", tags, values).SetConfig(&data.FieldConfig{DisplayNameFromDS: name}))
frame.Meta = &data.FrameMeta{
Custom: map[string]PrometheusQueryType{
"queryType": queryType,
},
}
frames = append(frames, frame)
}
return frames
}
func scalarToDataFrames(scalar *model.Scalar) data.Frames {
func scalarToDataFrames(scalar *model.Scalar, query *PrometheusQuery, queryType PrometheusQueryType) data.Frames {
timeVector := []time.Time{time.Unix(scalar.Timestamp.Unix(), 0).UTC()}
values := []float64{float64(scalar.Value)}
name := fmt.Sprintf("%g", values[0])
frames := data.Frames{data.NewFrame(name,
frame := data.NewFrame(name,
data.NewField("Time", nil, timeVector),
data.NewField("Value", nil, values).SetConfig(&data.FieldConfig{DisplayNameFromDS: name}))}
data.NewField("Value", nil, values).SetConfig(&data.FieldConfig{DisplayNameFromDS: name}))
frame.Meta = &data.FrameMeta{
Custom: map[string]PrometheusQueryType{
"queryType": queryType,
},
}
frames := data.Frames{frame}
return frames
}
func vectorToDataFrames(vector model.Vector, query *PrometheusQuery) data.Frames {
func vectorToDataFrames(vector model.Vector, query *PrometheusQuery, queryType PrometheusQueryType) data.Frames {
frames := data.Frames{}
for _, v := range vector {
name := formatLegend(v.Metric, query)
@@ -435,9 +420,15 @@ func vectorToDataFrames(vector model.Vector, query *PrometheusQuery) data.Frames
for k, v := range v.Metric {
tags[string(k)] = string(v)
}
frames = append(frames, data.NewFrame(name,
frame := data.NewFrame(name,
data.NewField("Time", nil, timeVector),
data.NewField("Value", tags, values).SetConfig(&data.FieldConfig{DisplayNameFromDS: name})))
data.NewField("Value", tags, values).SetConfig(&data.FieldConfig{DisplayNameFromDS: name}))
frame.Meta = &data.FrameMeta{
Custom: map[string]PrometheusQueryType{
"queryType": queryType,
},
}
frames = append(frames, frame)
}
return frames
pkg/tsdb/prometheus/prometheus_test.go
+32 -6
View File
@@ -303,10 +303,32 @@ func TestPrometheus_parseQuery(t *testing.T) {
dsInfo := &DatasourceInfo{}
models, err := service.parseQuery(query, dsInfo)
require.NoError(t, err)
require.Equal(t, Range, models[0].QueryType)
require.Equal(t, true, models[0].RangeQuery)
})
t.Run("parsing query model of with default query type", func(t *testing.T) {
t.Run("parsing query model of range and instant query", func(t *testing.T) {
timeRange := backend.TimeRange{
From: now,
To: now.Add(48 * time.Hour),
}
query := queryContext(`{
"expr": "go_goroutines",
"format": "time_series",
"intervalFactor": 1,
"refId": "A",
"range": true,
"instant": true,
}`, timeRange)
dsInfo := &DatasourceInfo{}
models, err := service.parseQuery(query, dsInfo)
require.NoError(t, err)
require.Equal(t, true, models[0].RangeQuery)
require.Equal(t, true, models[0].InstantQuery)
})
t.Run("parsing query model of with no query type", func(t *testing.T) {
timeRange := backend.TimeRange{
From: now,
To: now.Add(48 * time.Hour),
@@ -322,7 +344,7 @@ func TestPrometheus_parseQuery(t *testing.T) {
dsInfo := &DatasourceInfo{}
models, err := service.parseQuery(query, dsInfo)
require.NoError(t, err)
require.Equal(t, Range, models[0].QueryType)
require.Equal(t, true, models[0].RangeQuery)
})
}
@@ -335,7 +357,8 @@ func TestPrometheus_parseResponse(t *testing.T) {
{Value: 4, Timestamp: 4000},
{Value: 5, Timestamp: 5000},
}
value := p.Matrix{
value := make(map[PrometheusQueryType]p.Value)
value[Range] = p.Matrix{
&p.SampleStream{
Metric: p.Metric{"app": "Application", "tag2": "tag2"},
Values: values,
@@ -363,7 +386,8 @@ func TestPrometheus_parseResponse(t *testing.T) {
})
t.Run("vector response should be parsed normally", func(t *testing.T) {
value := p.Vector{
value := make(map[PrometheusQueryType]p.Value)
value[Range] = p.Vector{
&p.Sample{
Metric: p.Metric{"app": "Application", "tag2": "tag2"},
Value: 1,
@@ -392,10 +416,12 @@ func TestPrometheus_parseResponse(t *testing.T) {
})
t.Run("scalar response should be parsed normally", func(t *testing.T) {
value := &p.Scalar{
value := make(map[PrometheusQueryType]p.Value)
value[Range] = &p.Scalar{
Value: 1,
Timestamp: 1000,
}
query := &PrometheusQuery{}
res, err := parseResponse(value, query)
require.NoError(t, err)
pkg/tsdb/prometheus/types.go
+29 -4
View File
@@ -1,6 +1,18 @@
package prometheus
import "time"
import (
"time"
sdkhttpclient "github.com/grafana/grafana-plugin-sdk-go/backend/httpclient"
)
type DatasourceInfo struct {
ID int64
HTTPClientOpts sdkhttpclient.Options
URL string
HTTPMethod string
TimeInterval string
}
type PrometheusQuery struct {
Expr string
@@ -9,13 +21,26 @@ type PrometheusQuery struct {
Start time.Time
End time.Time
RefId string
QueryType PrometheusQueryType
InstantQuery bool
RangeQuery bool
UtcOffsetSec int64
}
type QueryModel struct {
Expr string `json:"expr"`
LegendFormat string `json:"legendFormat"`
Interval string `json:"interval"`
IntervalMS int64 `json:"intervalMS"`
StepMode string `json:"stepMode"`
RangeQuery bool `json:"range"`
InstantQuery bool `json:"instant"`
IntervalFactor int64 `json:"intervalFactor"`
UtcOffsetSec int64 `json:"utcOffsetSec"`
}
type PrometheusQueryType string
const (
Range PrometheusQueryType = "range"
//This is currently not used, but we will use it in next iteration
Range PrometheusQueryType = "range"
Instant PrometheusQueryType = "instant"
)
public/app/plugins/datasource/prometheus/datasource.test.ts
-48
View File
@@ -2110,54 +2110,6 @@ describe('prepareTargets', () => {
});
});
describe('prepareOptionsV2', () => {
it('creates instant target', () => {
const query: PromQuery = { refId: 'A', expr: 'go_goroutines', instant: true, format: 'table' };
const instanceSettings = ({ jsonData: {} } as unknown) as DataSourceInstanceSettings<PromOptions>;
const ds = new PrometheusDatasource(instanceSettings, templateSrvStub as any, timeSrvStub as any);
const instantTarget = ds.createTargetV2(query, 'instant');
expect(instantTarget.refId).toEqual('A');
expect(instantTarget.range).toEqual(false);
expect(instantTarget.instant).toEqual(true);
expect(instantTarget.format).toEqual('table');
});
it('creates range target', () => {
const query: PromQuery = { refId: 'A', expr: 'go_goroutines', range: true, format: 'time_series' };
const instanceSettings = ({ jsonData: {} } as unknown) as DataSourceInstanceSettings<PromOptions>;
const ds = new PrometheusDatasource(instanceSettings, templateSrvStub as any, timeSrvStub as any);
const rangeTarget = ds.createTargetV2(query, 'range');
expect(rangeTarget.refId).toEqual('A');
expect(rangeTarget.range).toEqual(true);
expect(rangeTarget.instant).toEqual(false);
expect(rangeTarget.format).toEqual('time_series');
});
it('creates both targets', () => {
const query: PromQuery = { refId: 'A', expr: 'go_goroutines', range: true, instant: true };
const instanceSettings = ({ jsonData: {} } as unknown) as DataSourceInstanceSettings<PromOptions>;
const ds = new PrometheusDatasource(instanceSettings, templateSrvStub as any, timeSrvStub as any);
const instantTarget = ds.createTargetV2(query, 'instant', 'table', true);
const rangeTarget = ds.createTargetV2(query, 'range', 'time_series', true);
expect(instantTarget.refId).toEqual('A_instant');
expect(rangeTarget.refId).toEqual('A');
expect(instantTarget.range).toEqual(false);
expect(rangeTarget.range).toEqual(true);
expect(instantTarget.instant).toEqual(true);
expect(rangeTarget.instant).toEqual(false);
expect(instantTarget.format).toEqual('table');
expect(rangeTarget.format).toEqual('time_series');
});
it('overrides the format', () => {
const query: PromQuery = { refId: 'A', expr: 'go_goroutines', range: true, format: 'time_series' };
const instanceSettings = ({ jsonData: {} } as unknown) as DataSourceInstanceSettings<PromOptions>;
const ds = new PrometheusDatasource(instanceSettings, templateSrvStub as any, timeSrvStub as any);
const rangeTarget = ds.createTargetV2(query, 'range', 'table');
expect(rangeTarget.format).toEqual('table');
});
});
describe('modifyQuery', () => {
describe('when called with ADD_FILTER', () => {
describe('and query has no labels', () => {
public/app/plugins/datasource/prometheus/datasource.ts
+1 -40
View File
@@ -289,52 +289,13 @@ export class PrometheusDatasource extends DataSourceWithBackend<PromQuery, PromO
};
};
createTargetV2 = (
target: PromQuery,
type: 'instant' | 'range',
format?: 'table' | 'time_series',
multipleQueryTypes?: boolean
): PromQuery => {
const refId = multipleQueryTypes && type !== 'range' ? `${target.refId}_${type}` : target.refId;
const resultFormat = format ?? target.format;
return {
...target,
range: type === 'range' ? true : false,
instant: type === 'instant' ? true : false,
utcOffsetSec: this.timeSrv.timeRange().to.utcOffset() * 60,
format: resultFormat,
refId,
};
};
prepareOptionsV2 = (options: DataQueryRequest<PromQuery>) => {
const targets: PromQuery[] = [];
options.targets.forEach((target) => {
if (target.range) {
const targetRange = this.createTargetV2(target, 'range', 'time_series');
targets.push(targetRange);
}
if (target.instant) {
const multipleQueryTypes = target.range && target.instant;
const targetInstant = this.createTargetV2(target, 'instant', 'table', multipleQueryTypes);
targets.push(targetInstant);
}
});
return { ...options, targets };
};
query(options: DataQueryRequest<PromQuery>): Observable<DataQueryResponse> {
// WIP - currently we want to run trough backend only if all queries are explore + range/instant queries
const shouldRunBackendQuery =
this.access === 'proxy' && options.app === CoreApp.Explore && !options.targets.some((query) => query.exemplar);
if (shouldRunBackendQuery) {
const newOptions = this.prepareOptionsV2(options);
return super.query(newOptions).pipe(map((response) => transformV2(response, newOptions)));
return super.query(options).pipe(map((response) => transformV2(response, options)));
// Run queries trough browser/proxy
} else {
const start = this.getPrometheusTime(options.range.from, false);
public/app/plugins/datasource/prometheus/result_transformer.ts
+17 -3
View File
@@ -16,6 +16,7 @@ import {
DataQueryResponse,
DataQueryRequest,
PreferredVisualisationType,
CoreApp,
} from '@grafana/data';
import { FetchResponse, getDataSourceSrv, getTemplateSrv } from '@grafana/runtime';
import { partition } from 'lodash';
@@ -41,12 +42,25 @@ interface TimeAndValue {
[TIME_SERIES_VALUE_FIELD_NAME]: number;
}
const isTableResult = (dataFrame: DataFrame, options: DataQueryRequest<PromQuery>): boolean => {
// We want to process instant results in Explore as table
if ((options.app === CoreApp.Explore && dataFrame.meta?.custom?.queryType) === 'instant') {
return true;
}
// We want to process all dataFrames with target.format === 'table' as table
const target = options.targets.find((target) => target.refId === dataFrame.refId);
if (target?.format === 'table') {
return true;
}
return false;
};
// V2 result trasnformer used to transform query results from queries that were run trough prometheus backend
export function transformV2(response: DataQueryResponse, options: DataQueryRequest<PromQuery>) {
// Get refIds that have table format as we need to process those to table reuslts
const tableRefIds = options.targets.filter((target) => target.format === 'table').map((target) => target.refId);
const [tableResults, otherResults]: [DataFrame[], DataFrame[]] = partition(response.data, (dataFrame) =>
dataFrame.refId ? tableRefIds.includes(dataFrame.refId) : false
isTableResult(dataFrame, options)
);
// For table results, we need to transform data frames to table data frames