From 46efe41e33983a4cfdd5a3391ba251816899e0e3 Mon Sep 17 00:00:00 2001 From: Kyle Brandt Date: Wed, 17 Apr 2024 13:53:38 -0400 Subject: [PATCH] Prometheus: (Instrumentation) Add rawExpr (pre-interpolation) to traces (#86449) --- pkg/promlib/models/query.go | 32 +++++++++++++++- pkg/promlib/models/query_test.go | 65 ++++++++++++++++++-------------- pkg/promlib/querydata/request.go | 37 +++++++++--------- 3 files changed, 85 insertions(+), 49 deletions(-) diff --git a/pkg/promlib/models/query.go b/pkg/promlib/models/query.go index 4f657183cec..930de98ab0c 100644 --- a/pkg/promlib/models/query.go +++ b/pkg/promlib/models/query.go @@ -3,6 +3,7 @@ package models import ( "embed" "encoding/json" + "fmt" "math" "strconv" "strings" @@ -12,6 +13,8 @@ import ( "github.com/grafana/grafana-plugin-sdk-go/backend/gtime" sdkapi "github.com/grafana/grafana-plugin-sdk-go/experimental/apis/data/v0alpha1" "github.com/prometheus/prometheus/model/labels" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" "github.com/grafana/grafana/pkg/promlib/intervalv2" ) @@ -169,11 +172,12 @@ type Scope struct { Matchers []*labels.Matcher } -func Parse(query backend.DataQuery, dsScrapeInterval string, intervalCalculator intervalv2.Calculator, fromAlert bool, enableScope bool) (*Query, error) { +func Parse(span trace.Span, query backend.DataQuery, dsScrapeInterval string, intervalCalculator intervalv2.Calculator, fromAlert bool, enableScope bool) (*Query, error) { model := &QueryModel{} if err := json.Unmarshal(query.JSON, model); err != nil { return nil, err } + span.SetAttributes(attribute.String("rawExpr", model.Expr)) // Final step value for prometheus calculatedStep, err := calculatePrometheusInterval(model.Interval, dsScrapeInterval, int64(model.IntervalMS), model.IntervalFactor, query, intervalCalculator) @@ -198,6 +202,26 @@ func Parse(query backend.DataQuery, dsScrapeInterval string, intervalCalculator scopeFilters = model.Scope.Filters } + if len(scopeFilters) > 0 { + span.SetAttributes(attribute.StringSlice("scopeFilters", func() []string { + var filters []string + for _, f := range scopeFilters { + filters = append(filters, fmt.Sprintf("%q %q %q", f.Key, f.Operator, f.Value)) + } + return filters + }())) + } + + if len(model.AdhocFilters) > 0 { + span.SetAttributes(attribute.StringSlice("adhocFilters", func() []string { + var filters []string + for _, f := range model.AdhocFilters { + filters = append(filters, fmt.Sprintf("%q %q %q", f.Key, f.Operator, f.Value)) + } + return filters + }())) + } + expr, err = ApplyQueryFilters(expr, scopeFilters, model.AdhocFilters) if err != nil { return nil, err @@ -214,6 +238,12 @@ func Parse(query backend.DataQuery, dsScrapeInterval string, intervalCalculator model.Exemplar = false } + span.SetAttributes( + attribute.String("expr", expr), + attribute.Int64("start_unixnano", query.TimeRange.From.UnixNano()), + attribute.Int64("stop_unixnano", query.TimeRange.To.UnixNano()), + ) + return &Query{ Expr: expr, Step: calculatedStep, diff --git a/pkg/promlib/models/query_test.go b/pkg/promlib/models/query_test.go index 3bba6eb78d9..21925581f53 100644 --- a/pkg/promlib/models/query_test.go +++ b/pkg/promlib/models/query_test.go @@ -1,6 +1,7 @@ package models_test import ( + "context" "fmt" "reflect" "testing" @@ -10,6 +11,7 @@ import ( sdkapi "github.com/grafana/grafana-plugin-sdk-go/experimental/apis/data/v0alpha1" "github.com/grafana/grafana-plugin-sdk-go/experimental/schemabuilder" "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel" "github.com/grafana/grafana/pkg/promlib/intervalv2" "github.com/grafana/grafana/pkg/promlib/models" @@ -18,9 +20,12 @@ import ( var ( now = time.Now() intervalCalculator = intervalv2.NewCalculator() + tracer = otel.Tracer("instrumentation/package/name") ) func TestParse(t *testing.T) { + _, span := tracer.Start(context.Background(), "operation") + defer span.End() t.Run("parsing query from unified alerting", func(t *testing.T) { timeRange := backend.TimeRange{ From: now, @@ -39,7 +44,7 @@ func TestParse(t *testing.T) { RefID: "A", } - res, err := models.Parse(q, "15s", intervalCalculator, true, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, true, false) require.NoError(t, err) require.Equal(t, false, res.ExemplarQuery) }) @@ -56,7 +61,7 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, time.Duration(1)*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, time.Second*30, res.Step) }) @@ -74,7 +79,7 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, time.Duration(1)*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, time.Second*15, res.Step) }) @@ -92,7 +97,7 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, time.Duration(1)*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, time.Minute*20, res.Step) }) @@ -110,7 +115,7 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, time.Duration(1)*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, time.Minute*2, res.Step) }) @@ -128,7 +133,7 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, time.Duration(1)*time.Minute) - res, err := models.Parse(q, "240s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "240s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, time.Minute*4, res.Step) }) @@ -147,7 +152,7 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, time.Duration(1)*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, "rate(ALERTS{job=\"test\" [2m]})", res.Expr) require.Equal(t, 120*time.Second, res.Step) @@ -168,7 +173,7 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, time.Duration(1)*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, "rate(ALERTS{job=\"test\" [2m]})", res.Expr) }) @@ -187,7 +192,7 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, time.Duration(1)*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, "rate(ALERTS{job=\"test\" [120000]})", res.Expr) }) @@ -206,7 +211,7 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, time.Duration(1)*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, "rate(ALERTS{job=\"test\" [120000]}) + rate(ALERTS{job=\"test\" [2m]})", res.Expr) }) @@ -225,7 +230,7 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, time.Duration(1)*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, "rate(ALERTS{job=\"test\" [120000]}) + rate(ALERTS{job=\"test\" [2m]})", res.Expr) }) @@ -243,7 +248,7 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, time.Duration(1)*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, "rate(ALERTS{job=\"test\" [172800s]})", res.Expr) }) @@ -261,7 +266,7 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, time.Duration(1)*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, "rate(ALERTS{job=\"test\" [172800]})", res.Expr) }) @@ -279,7 +284,7 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, time.Duration(1)*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, "rate(ALERTS{job=\"test\" [172800s]})", res.Expr) }) @@ -297,7 +302,7 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, time.Duration(1)*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, "rate(ALERTS{job=\"test\" [0]})", res.Expr) }) @@ -315,7 +320,7 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, time.Duration(1)*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, "rate(ALERTS{job=\"test\" [1]})", res.Expr) }) @@ -333,7 +338,7 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, time.Duration(1)*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, "rate(ALERTS{job=\"test\" [172800000]})", res.Expr) }) @@ -351,7 +356,7 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, time.Duration(1)*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, "rate(ALERTS{job=\"test\" [20]})", res.Expr) }) @@ -370,7 +375,7 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, time.Duration(1)*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, "rate(ALERTS{job=\"test\" [20m0s]})", res.Expr) }) @@ -389,7 +394,7 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, 1*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, "rate(ALERTS{job=\"test\" [1m0s]})", res.Expr) require.Equal(t, 1*time.Minute, res.Step) @@ -408,7 +413,7 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, 2*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, "rate(ALERTS{job=\"test\" [135000]})", res.Expr) }) @@ -426,7 +431,7 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, 2*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, "rate(ALERTS{job=\"test\" [135000]}) + rate(ALERTS{job=\"test\" [2m15s]})", res.Expr) }) @@ -444,7 +449,7 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, 2*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, "rate(ALERTS{job=\"test\" [135000]}) + rate(ALERTS{job=\"test\" [2m15s]})", res.Expr) }) @@ -463,7 +468,7 @@ func TestParse(t *testing.T) { "range": true }`, timeRange, time.Duration(1)*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, true, res.RangeQuery) }) @@ -483,7 +488,7 @@ func TestParse(t *testing.T) { "instant": true }`, timeRange, time.Duration(1)*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, true, res.RangeQuery) require.Equal(t, true, res.InstantQuery) @@ -502,13 +507,15 @@ func TestParse(t *testing.T) { "refId": "A" }`, timeRange, time.Duration(1)*time.Minute) - res, err := models.Parse(q, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, q, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, true, res.RangeQuery) }) } func TestRateInterval(t *testing.T) { + _, span := tracer.Start(context.Background(), "operation") + defer span.End() type args struct { expr string interval string @@ -633,7 +640,7 @@ func TestRateInterval(t *testing.T) { t.Run(tt.name, func(t *testing.T) { q := mockQuery(tt.args.expr, tt.args.interval, tt.args.intervalMs, tt.args.timeRange) q.MaxDataPoints = 12384 - res, err := models.Parse(q, tt.args.dsScrapeInterval, intervalCalculator, false, false) + res, err := models.Parse(span, q, tt.args.dsScrapeInterval, intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, tt.want.Expr, res.Expr) require.Equal(t, tt.want.Step, res.Step) @@ -668,7 +675,7 @@ func TestRateInterval(t *testing.T) { "utcOffsetSec":3600 }`), } - res, err := models.Parse(query, "30s", intervalCalculator, false, false) + res, err := models.Parse(span, query, "30s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, "sum(rate(process_cpu_seconds_total[2m0s]))", res.Expr) require.Equal(t, 30*time.Second, res.Step) @@ -703,7 +710,7 @@ func TestRateInterval(t *testing.T) { "maxDataPoints": 1055 }`), } - res, err := models.Parse(query, "15s", intervalCalculator, false, false) + res, err := models.Parse(span, query, "15s", intervalCalculator, false, false) require.NoError(t, err) require.Equal(t, "sum(rate(cache_requests_total[1m0s]))", res.Expr) require.Equal(t, 15*time.Second, res.Step) diff --git a/pkg/promlib/querydata/request.go b/pkg/promlib/querydata/request.go index aabd347b9ae..b9afb827111 100644 --- a/pkg/promlib/querydata/request.go +++ b/pkg/promlib/querydata/request.go @@ -10,7 +10,6 @@ import ( "github.com/grafana/grafana-plugin-sdk-go/backend" "github.com/grafana/grafana-plugin-sdk-go/data" "github.com/grafana/grafana-plugin-sdk-go/data/utils/maputil" - "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/trace" "github.com/grafana/grafana-plugin-sdk-go/backend/log" @@ -94,14 +93,8 @@ func (s *QueryData) Execute(ctx context.Context, req *backend.QueryDataRequest) hasPrometheusDataplaneFeatureFlag := cfg.FeatureToggles().IsEnabled("prometheusDataplane") for _, q := range req.Queries { - query, err := models.Parse(q, s.TimeInterval, s.intervalCalculator, fromAlert, hasPromQLScopeFeatureFlag) - if err != nil { - return &result, err - } - - r := s.fetch(ctx, s.client, query, hasPrometheusDataplaneFeatureFlag) + r := s.handleQuery(ctx, q, fromAlert, hasPromQLScopeFeatureFlag, hasPrometheusDataplaneFeatureFlag) if r == nil { - s.log.FromContext(ctx).Debug("Received nil response from runQuery", "query", query.Expr) continue } result.Responses[q.RefID] = *r @@ -110,10 +103,24 @@ func (s *QueryData) Execute(ctx context.Context, req *backend.QueryDataRequest) return &result, nil } -func (s *QueryData) fetch(ctx context.Context, client *client.Client, q *models.Query, enablePrometheusDataplane bool) *backend.DataResponse { - traceCtx, end := s.trace(ctx, q) - defer end() +func (s *QueryData) handleQuery(ctx context.Context, bq backend.DataQuery, fromAlert, hasPromQLScopeFeatureFlag, hasPrometheusDataplaneFeatureFlag bool) *backend.DataResponse { + traceCtx, span := s.tracer.Start(ctx, "datasource.prometheus") + defer span.End() + query, err := models.Parse(span, bq, s.TimeInterval, s.intervalCalculator, fromAlert, hasPromQLScopeFeatureFlag) + if err != nil { + return &backend.DataResponse{ + Error: err, + } + } + r := s.fetch(traceCtx, s.client, query, hasPrometheusDataplaneFeatureFlag) + if r == nil { + s.log.FromContext(ctx).Debug("Received nil response from runQuery", "query", query.Expr) + } + return r +} + +func (s *QueryData) fetch(traceCtx context.Context, client *client.Client, q *models.Query, enablePrometheusDataplane bool) *backend.DataResponse { logger := s.log.FromContext(traceCtx) logger.Debug("Sending query", "start", q.Start, "end", q.End, "step", q.Step, "query", q.Expr) @@ -218,11 +225,3 @@ func (s *QueryData) exemplarQuery(ctx context.Context, c *client.Client, q *mode }() return s.parseResponse(ctx, q, res, enablePrometheusDataplaneFlag) } - -func (s *QueryData) trace(ctx context.Context, q *models.Query) (context.Context, func()) { - return utils.StartTrace(ctx, s.tracer, "datasource.prometheus", - attribute.String("expr", q.Expr), - attribute.Int64("start_unixnano", q.Start.UnixNano()), - attribute.Int64("stop_unixnano", q.End.UnixNano()), - ) -}