Prometheus: Run Explore both queries trough backend (#39462)

* Prometheus: Run Explore both queries trough backend

* Refactor, simplify

* Set default values for query type selector

* Run multiple queries as one query trough backend

* Remove trailing newlines

* Pass utcOffset

* Remove trailing comma

* Add meta to frames only at 1 place

* Set exemplars to falsee if they are disabled

* Revert "Set exemplars to falsee if they are disabled"

This reverts commit e7b697c9f3.
This commit is contained in:
Ivana Huckova
2021-09-29 16:42:31 +02:00
committed by GitHub
parent aeeaa67b0a
commit df81676d9e
6 changed files with 162 additions and 129 deletions

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
@@ -154,13 +134,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)
@@ -168,30 +142,38 @@ 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)
response[Range] = rangeResponse
}
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)
}
default:
return &result, fmt.Errorf("unknown Query type detected %#v", query.QueryType)
response[Instant] = instantResponse
}
frame, err := parseResponse(response, query)
frames, err := parseResponse(response, query)
if err != nil {
return &result, err
}
result.Responses[query.RefId] = backend.DataResponse{
Frames: frame,
Frames: frames,
}
}
@@ -309,61 +291,60 @@ 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) {
frames := data.Frames{}
func parseResponse(value map[PrometheusQueryType]model.Value, query *PrometheusQuery) (data.Frames, error) {
allFrames := data.Frames{}
for queryType, value := range value {
var frames data.Frames
matrix, ok := value.(model.Matrix)
if ok {
matrixFrames := matrixToDataFrames(matrix, query)
frames = append(frames, matrixFrames...)
frames = matrixToDataFrames(matrix, query, queryType)
}
vector, ok := value.(model.Vector)
if ok {
vectorFrames := vectorToDataFrames(vector, query)
frames = append(frames, vectorFrames...)
frames = vectorToDataFrames(vector, query, queryType)
}
scalar, ok := value.(*model.Scalar)
if ok {
scalarFrames := scalarToDataFrames(scalar)
frames = append(frames, scalarFrames...)
frames = scalarToDataFrames(scalar, query, queryType)
}
return frames, nil
for _, frame := range frames {
frame.Meta = &data.FrameMeta{
Custom: map[string]PrometheusQueryType{
"queryType": queryType,
},
}
}
allFrames = append(allFrames, frames...)
}
return allFrames, nil
}
// IsAPIError returns whether err is or wraps a Prometheus error.
@@ -396,7 +377,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 {
@@ -411,26 +392,26 @@ 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}))
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}))
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)
@@ -440,9 +421,10 @@ 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}))
frames = append(frames, frame)
}
return frames

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)

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
Instant PrometheusQueryType = "instant"
)

View File

@@ -18,6 +18,10 @@ export const PromExploreQueryEditor: FC<Props> = (props: Props) => {
if (query.exemplar === undefined) {
onChange({ ...query, exemplar: true });
}
if (!query.instant && !query.range) {
onChange({ ...query, instant: true, range: true });
}
}, [onChange, query]);
function onChangeQueryStep(value: string) {

View File

@@ -289,36 +289,18 @@ export class PrometheusDatasource extends DataSourceWithBackend<PromQuery, PromO
};
};
prepareOptionsV2 = (options: DataQueryRequest<PromQuery>) => {
const targets = options.targets.map((target) => {
//This is currently only preparing options for Explore queries where we know the format of data we want to receive
if (target.instant) {
return { ...target, instant: true, range: false, format: 'table' };
}
return {
...target,
instant: false,
range: true,
format: 'time_series',
utcOffsetSec: this.timeSrv.timeRange().to.utcOffset() * 60,
};
});
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) &&
// When running both queries, run through proxy
!options.targets.some((query) => query.instant && query.range);
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)));
const targets = options.targets.map((target) => ({
...target,
// We need to pass utcOffsetSec to backend to calculate aligned range
utcOffsetSec: this.timeSrv.timeRange().to.utcOffset() * 60,
}));
return super.query({ ...options, targets }).pipe(map((response) => transformV2(response, options)));
// Run queries trough browser/proxy
} else {
const start = this.getPrometheusTime(options.range.from, false);

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