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

Prometheus: Run exemplars explore queries through backend (#39531)

Browse Source 
* 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

* WIP: Implementatioon of exemplars

* add sampling for exemplars

* Refactor to use response as custom metadata

* Simplify processing of exemplars

* Update, clean up

* Refactor the way how we get available exemplars

* Simplify exemplars disabling and running on frontend

* Add tests

* Update toggle

* Remove console log

* Fix go linting

* Fix e2e test

* Trigger Build

* Compare lengts, small fix

* Remove duplicated time check

* Address feedback

* Remove redundant ! as not needed

* Update
This commit is contained in:
Ivana Huckova
2021-10-12 13:16:09 +02:00
committed by GitHub
parent 2f0eccb421
commit 19ad08e6b8
18 changed files with 893 additions and 925 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/tsdb/prometheus/prometheus.go
+198 -38
View File
@@ -8,6 +8,7 @@ import (
"math"
"net/http"
"regexp"
"sort"
"strconv"
"strings"
"time"
@@ -142,16 +143,16 @@ func (s *Service) QueryData(ctx context.Context, req *backend.QueryDataRequest)
span.SetTag("stop_unixnano", query.End.UnixNano())
defer span.Finish()
response := make(map[PrometheusQueryType]model.Value)
response := make(map[PrometheusQueryType]interface{})
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),
}
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)
@@ -166,6 +167,15 @@ func (s *Service) QueryData(ctx context.Context, req *backend.QueryDataRequest)
}
response[Instant] = instantResponse
}
// For now, we ignore exemplar errors and continue with processing of other results
if query.ExemplarQuery {
exemplarResponse, err := client.QueryExemplars(ctx, query.Expr, timeRange.Start, timeRange.End)
if err != nil {
exemplarResponse = nil
plog.Error("Exemplar query", query.Expr, "failed with", err)
}
response[Exemplar] = exemplarResponse
}
frames, err := parseResponse(response, query)
if err != nil {
@@ -299,53 +309,54 @@ func (s *Service) parseQuery(queryContext *backend.QueryDataRequest, dsInfo *Dat
}
qs = append(qs, &PrometheusQuery{
Expr: expr,
Step: interval,
LegendFormat: model.LegendFormat,
Start: query.TimeRange.From,
End: query.TimeRange.To,
RefId: query.RefID,
InstantQuery: model.InstantQuery,
RangeQuery: rangeQuery,
UtcOffsetSec: model.UtcOffsetSec,
Expr: expr,
Step: interval,
LegendFormat: model.LegendFormat,
Start: query.TimeRange.From,
End: query.TimeRange.To,
RefId: query.RefID,
InstantQuery: model.InstantQuery,
RangeQuery: rangeQuery,
ExemplarQuery: model.ExemplarQuery,
UtcOffsetSec: model.UtcOffsetSec,
})
}
return qs, nil
}
func parseResponse(value map[PrometheusQueryType]model.Value, query *PrometheusQuery) (data.Frames, error) {
allFrames := data.Frames{}
for queryType, value := range value {
var frames data.Frames
func parseResponse(value map[PrometheusQueryType]interface{}, query *PrometheusQuery) (data.Frames, error) {
frames := data.Frames{}
for _, value := range value {
matrix, ok := value.(model.Matrix)
if ok {
frames = matrixToDataFrames(matrix, query, queryType)
matrixFrames := matrixToDataFrames(matrix, query)
frames = append(frames, matrixFrames...)
continue
}
vector, ok := value.(model.Vector)
if ok {
frames = vectorToDataFrames(vector, query, queryType)
vectorFrames := vectorToDataFrames(vector, query)
frames = append(frames, vectorFrames...)
continue
}
scalar, ok := value.(*model.Scalar)
if ok {
frames = scalarToDataFrames(scalar, query, queryType)
scalarFrames := scalarToDataFrames(scalar, query)
frames = append(frames, scalarFrames...)
continue
}
for _, frame := range frames {
frame.Meta = &data.FrameMeta{
Custom: map[string]PrometheusQueryType{
"queryType": queryType,
},
}
exemplar, ok := value.([]apiv1.ExemplarQueryResult)
if ok {
exemplarFrames := exemplarToDataFrames(exemplar, query)
frames = append(frames, exemplarFrames...)
continue
}
allFrames = append(allFrames, frames...)
}
return allFrames, nil
return frames, nil
}
// IsAPIError returns whether err is or wraps a Prometheus error.
@@ -378,7 +389,7 @@ func calculateRateInterval(interval time.Duration, scrapeInterval string, interv
return rateInterval
}
func matrixToDataFrames(matrix model.Matrix, query *PrometheusQuery, queryType PrometheusQueryType) data.Frames {
func matrixToDataFrames(matrix model.Matrix, query *PrometheusQuery) data.Frames {
frames := data.Frames{}
for _, v := range matrix {
@@ -396,23 +407,34 @@ func matrixToDataFrames(matrix model.Matrix, query *PrometheusQuery, queryType P
frame := data.NewFrame(name,
data.NewField("Time", nil, timeVector),
data.NewField("Value", tags, values).SetConfig(&data.FieldConfig{DisplayNameFromDS: name}))
frame.Meta = &data.FrameMeta{
Custom: map[string]string{
"resultType": "matrix",
},
}
frames = append(frames, frame)
}
return frames
}
func scalarToDataFrames(scalar *model.Scalar, query *PrometheusQuery, queryType PrometheusQueryType) data.Frames {
func scalarToDataFrames(scalar *model.Scalar, query *PrometheusQuery) data.Frames {
timeVector := []time.Time{time.Unix(scalar.Timestamp.Unix(), 0).UTC()}
values := []float64{float64(scalar.Value)}
name := fmt.Sprintf("%g", values[0])
frame := data.NewFrame(name,
data.NewField("Time", nil, timeVector),
data.NewField("Value", nil, values).SetConfig(&data.FieldConfig{DisplayNameFromDS: name}))
frame.Meta = &data.FrameMeta{
Custom: map[string]string{
"resultType": "scalar",
},
}
frames := data.Frames{frame}
return frames
}
func vectorToDataFrames(vector model.Vector, query *PrometheusQuery, queryType PrometheusQueryType) data.Frames {
func vectorToDataFrames(vector model.Vector, query *PrometheusQuery) data.Frames {
frames := data.Frames{}
for _, v := range vector {
name := formatLegend(v.Metric, query)
@@ -425,8 +447,146 @@ func vectorToDataFrames(vector model.Vector, query *PrometheusQuery, queryType P
frame := data.NewFrame(name,
data.NewField("Time", nil, timeVector),
data.NewField("Value", tags, values).SetConfig(&data.FieldConfig{DisplayNameFromDS: name}))
frame.Meta = &data.FrameMeta{
Custom: map[string]string{
"resultType": "vector",
},
}
frames = append(frames, frame)
}
return frames
}
func exemplarToDataFrames(response []apiv1.ExemplarQueryResult, query *PrometheusQuery) data.Frames {
frames := data.Frames{}
events := make([]ExemplarEvent, 0)
for _, exemplarData := range response {
for _, exemplar := range exemplarData.Exemplars {
event := ExemplarEvent{}
exemplarTime := time.Unix(exemplar.Timestamp.Unix(), 0).UTC()
event.Time = exemplarTime
event.Value = float64(exemplar.Value)
event.Labels = make(map[string]string)
for label, value := range exemplar.Labels {
event.Labels[string(label)] = string(value)
}
for seriesLabel, seriesValue := range exemplarData.SeriesLabels {
event.Labels[string(seriesLabel)] = string(seriesValue)
}
events = append(events, event)
}
}
//Sampling of exemplars
bucketedExemplars := make(map[string][]ExemplarEvent)
values := make([]float64, 0)
//Create bucketed exemplars based on aligned timestamp
for _, event := range events {
alignedTs := fmt.Sprintf("%.0f", math.Floor(float64(event.Time.Unix())/query.Step.Seconds())*query.Step.Seconds())
_, ok := bucketedExemplars[alignedTs]
if !ok {
bucketedExemplars[alignedTs] = make([]ExemplarEvent, 0)
}
bucketedExemplars[alignedTs] = append(bucketedExemplars[alignedTs], event)
values = append(values, event.Value)
}
//Calculate standard deviation
standardDeviation := deviation(values)
//Create slice with all of the bucketed exemplars
sampledBuckets := make([]string, len(bucketedExemplars))
for bucketTimes := range bucketedExemplars {
sampledBuckets = append(sampledBuckets, bucketTimes)
}
sort.Strings(sampledBuckets)
//Sample exemplars based ona value, so we are not showing too many of them
sampleExemplars := make([]ExemplarEvent, 0)
for _, bucket := range sampledBuckets {
exemplarsInBucket := bucketedExemplars[bucket]
if len(exemplarsInBucket) == 1 {
sampleExemplars = append(sampleExemplars, exemplarsInBucket[0])
} else {
bucketValues := make([]float64, len(exemplarsInBucket))
for _, exemplar := range exemplarsInBucket {
bucketValues = append(bucketValues, exemplar.Value)
}
sort.Slice(bucketValues, func(i, j int) bool {
return bucketValues[i] > bucketValues[j]
})
sampledBucketValues := make([]float64, 0)
for _, value := range bucketValues {
if len(sampledBucketValues) == 0 {
sampledBucketValues = append(sampledBucketValues, value)
} else {
// Then take values only when at least 2 standard deviation distance to previously taken value
prev := sampledBucketValues[len(sampledBucketValues)-1]
if standardDeviation != 0 && prev-value >= float64(2)*standardDeviation {
sampledBucketValues = append(sampledBucketValues, value)
}
}
}
for _, valueBucket := range sampledBucketValues {
for _, exemplar := range exemplarsInBucket {
if exemplar.Value == valueBucket {
sampleExemplars = append(sampleExemplars, exemplar)
}
}
}
}
}
// Create DF from sampled exemplars
timeVector := make([]time.Time, 0, len(sampleExemplars))
valuesVector := make([]float64, 0, len(sampleExemplars))
labelsVector := make(map[string][]string, len(sampleExemplars))
for _, exemplar := range sampleExemplars {
timeVector = append(timeVector, exemplar.Time)
valuesVector = append(valuesVector, exemplar.Value)
for label, value := range exemplar.Labels {
if labelsVector[label] == nil {
labelsVector[label] = make([]string, 0)
}
labelsVector[label] = append(labelsVector[label], value)
}
}
frame := data.NewFrame("exemplar",
data.NewField("Time", nil, timeVector),
data.NewField("Value", nil, valuesVector))
for label, vector := range labelsVector {
frame.Fields = append(frame.Fields, data.NewField(label, nil, vector))
}
frame.Meta = &data.FrameMeta{
Custom: map[string]PrometheusQueryType{
"resultType": "exemplar",
},
}
frames = append(frames, frame)
return frames
}
func deviation(values []float64) float64 {
var sum, mean, sd float64
valuesLen := float64(len(values))
for _, value := range values {
sum += value
}
mean = sum / valuesLen
for j := 0; j < len(values); j++ {
sd += math.Pow(values[j]-mean, 2)
}
return math.Sqrt(sd / (valuesLen - 1))
}
pkg/tsdb/prometheus/prometheus_test.go
+59 -3
View File
@@ -6,6 +6,7 @@ import (
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana/pkg/tsdb/intervalv2"
apiv1 "github.com/prometheus/client_golang/api/prometheus/v1"
p "github.com/prometheus/common/model"
"github.com/stretchr/testify/require"
)
@@ -349,6 +350,60 @@ func TestPrometheus_parseQuery(t *testing.T) {
}
func TestPrometheus_parseResponse(t *testing.T) {
t.Run("exemplars response should be sampled and parsed normally", func(t *testing.T) {
value := make(map[PrometheusQueryType]interface{})
exemplars := []apiv1.ExemplarQueryResult{
{
SeriesLabels: p.LabelSet{
"__name__": "tns_request_duration_seconds_bucket",
"instance": "app:80",
"job": "tns/app",
},
Exemplars: []apiv1.Exemplar{
{
Labels: p.LabelSet{"traceID": "test1"},
Value: 0.003535405,
Timestamp: p.TimeFromUnixNano(time.Now().Add(-2 * time.Minute).UnixNano()),
},
{
Labels: p.LabelSet{"traceID": "test2"},
Value: 0.005555605,
Timestamp: p.TimeFromUnixNano(time.Now().Add(-4 * time.Minute).UnixNano()),
},
{
Labels: p.LabelSet{"traceID": "test3"},
Value: 0.007545445,
Timestamp: p.TimeFromUnixNano(time.Now().Add(-6 * time.Minute).UnixNano()),
},
{
Labels: p.LabelSet{"traceID": "test4"},
Value: 0.009545445,
Timestamp: p.TimeFromUnixNano(time.Now().Add(-7 * time.Minute).UnixNano()),
},
},
},
}
value[Exemplar] = exemplars
query := &PrometheusQuery{
LegendFormat: "legend {{app}}",
}
res, err := parseResponse(value, query)
require.NoError(t, err)
// Test fields
require.Len(t, res, 1)
require.Equal(t, res[0].Name, "exemplar")
require.Equal(t, res[0].Fields[0].Name, "Time")
require.Equal(t, res[0].Fields[1].Name, "Value")
require.Len(t, res[0].Fields, 6)
// Test correct values (sampled to 2)
require.Equal(t, res[0].Fields[1].Len(), 2)
require.Equal(t, res[0].Fields[1].At(0), 0.009545445)
require.Equal(t, res[0].Fields[1].At(1), 0.003535405)
})
t.Run("matrix response should be parsed normally", func(t *testing.T) {
values := []p.SamplePair{
{Value: 1, Timestamp: 1000},
@@ -357,7 +412,7 @@ func TestPrometheus_parseResponse(t *testing.T) {
{Value: 4, Timestamp: 4000},
{Value: 5, Timestamp: 5000},
}
value := make(map[PrometheusQueryType]p.Value)
value := make(map[PrometheusQueryType]interface{})
value[Range] = p.Matrix{
&p.SampleStream{
Metric: p.Metric{"app": "Application", "tag2": "tag2"},
@@ -386,7 +441,7 @@ func TestPrometheus_parseResponse(t *testing.T) {
})
t.Run("vector response should be parsed normally", func(t *testing.T) {
value := make(map[PrometheusQueryType]p.Value)
value := make(map[PrometheusQueryType]interface{})
value[Range] = p.Vector{
&p.Sample{
Metric: p.Metric{"app": "Application", "tag2": "tag2"},
@@ -405,6 +460,7 @@ func TestPrometheus_parseResponse(t *testing.T) {
require.Len(t, res[0].Fields, 2)
require.Len(t, res[0].Fields[0].Labels, 0)
require.Equal(t, res[0].Fields[0].Name, "Time")
require.Equal(t, res[0].Fields[0].Name, "Time")
require.Len(t, res[0].Fields[1].Labels, 2)
require.Equal(t, res[0].Fields[1].Labels.String(), "app=Application, tag2=tag2")
require.Equal(t, res[0].Fields[1].Name, "Value")
@@ -416,7 +472,7 @@ func TestPrometheus_parseResponse(t *testing.T) {
})
t.Run("scalar response should be parsed normally", func(t *testing.T) {
value := make(map[PrometheusQueryType]p.Value)
value := make(map[PrometheusQueryType]interface{})
value[Range] = &p.Scalar{
Value: 1,
Timestamp: 1000,
pkg/tsdb/prometheus/types.go
+20 -11
View File
@@ -15,15 +15,22 @@ type DatasourceInfo struct {
}
type PrometheusQuery struct {
Expr string
Step time.Duration
LegendFormat string
Start time.Time
End time.Time
RefId string
InstantQuery bool
RangeQuery bool
UtcOffsetSec int64
Expr string
Step time.Duration
LegendFormat string
Start time.Time
End time.Time
RefId string
InstantQuery bool
RangeQuery bool
ExemplarQuery bool
UtcOffsetSec int64
}
type ExemplarEvent struct {
Time time.Time
Value float64
Labels map[string]string
}
type QueryModel struct {
@@ -34,6 +41,7 @@ type QueryModel struct {
StepMode string `json:"stepMode"`
RangeQuery bool `json:"range"`
InstantQuery bool `json:"instant"`
ExemplarQuery bool `json:"exemplar"`
IntervalFactor int64 `json:"intervalFactor"`
UtcOffsetSec int64 `json:"utcOffsetSec"`
}
@@ -41,6 +49,7 @@ type QueryModel struct {
type PrometheusQueryType string
const (
Range PrometheusQueryType = "range"
Instant PrometheusQueryType = "instant"
Range PrometheusQueryType = "range"
Instant PrometheusQueryType = "instant"
Exemplar PrometheusQueryType = "exemplar"
)