From 30cb720da5f314afa83a8a25470bd5b8ec1fc221 Mon Sep 17 00:00:00 2001 From: Andre Pereira Date: Fri, 13 Oct 2023 13:13:51 +0100 Subject: [PATCH] Chore: Add tracing to tempo, parca and pyroscope datasource backends (#76368) * Added spans to trace.go * Added spans to search_stream.go * Added spans to parca datasource * Added spans for pyroscope * Fix tests * Fix another test * Lint * Revert "Fix another test" This reverts commit a1639049e3bce19342b6a68c0e9a6d78f0332c0e. * Use grafana-sdk-go tracing --- .../grafana-pyroscope-datasource/instance.go | 5 +++ .../pyroscopeClient.go | 24 ++++++++++++ .../grafana-pyroscope-datasource/query.go | 17 +++++++++ pkg/tsdb/parca/plugin.go | 6 +++ pkg/tsdb/parca/query.go | 13 +++++++ pkg/tsdb/parca/resources.go | 28 ++++++++++++++ pkg/tsdb/tempo/search_stream.go | 38 ++++++++++++++++--- pkg/tsdb/tempo/search_stream_test.go | 6 +-- pkg/tsdb/tempo/trace.go | 19 ++++++++++ 9 files changed, 148 insertions(+), 8 deletions(-) diff --git a/pkg/tsdb/grafana-pyroscope-datasource/instance.go b/pkg/tsdb/grafana-pyroscope-datasource/instance.go index 5ffd40e1ef21..35d20788370c 100644 --- a/pkg/tsdb/grafana-pyroscope-datasource/instance.go +++ b/pkg/tsdb/grafana-pyroscope-datasource/instance.go @@ -10,9 +10,12 @@ import ( "github.com/grafana/grafana-plugin-sdk-go/backend" "github.com/grafana/grafana-plugin-sdk-go/backend/instancemgmt" + "github.com/grafana/grafana-plugin-sdk-go/backend/tracing" "github.com/grafana/grafana-plugin-sdk-go/data" "github.com/grafana/grafana/pkg/infra/httpclient" "github.com/grafana/grafana/pkg/services/accesscontrol" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" ) var ( @@ -58,6 +61,8 @@ func NewPyroscopeDatasource(httpClientProvider httpclient.Provider, settings bac } func (d *PyroscopeDatasource) CallResource(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error { + ctx, span := tracing.DefaultTracer().Start(ctx, "datasource.pyroscope.CallResource", trace.WithAttributes(attribute.String("path", req.Path), attribute.String("method", req.Method))) + defer span.End() logger.Debug("CallResource", "Path", req.Path, "Method", req.Method, "Body", req.Body) if req.Path == "profileTypes" { return d.profileTypes(ctx, req, sender) diff --git a/pkg/tsdb/grafana-pyroscope-datasource/pyroscopeClient.go b/pkg/tsdb/grafana-pyroscope-datasource/pyroscopeClient.go index 3653c455ae2a..007ee6d06050 100644 --- a/pkg/tsdb/grafana-pyroscope-datasource/pyroscopeClient.go +++ b/pkg/tsdb/grafana-pyroscope-datasource/pyroscopeClient.go @@ -6,11 +6,15 @@ import ( "net/http" "strings" + "github.com/grafana/grafana-plugin-sdk-go/backend/tracing" typesv1 "github.com/grafana/pyroscope/api/gen/proto/go/types/v1" "github.com/bufbuild/connect-go" querierv1 "github.com/grafana/pyroscope/api/gen/proto/go/querier/v1" "github.com/grafana/pyroscope/api/gen/proto/go/querier/v1/querierv1connect" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/trace" ) type ProfileType struct { @@ -67,8 +71,12 @@ func NewPyroscopeClient(httpClient *http.Client, url string) *PyroscopeClient { } func (c *PyroscopeClient) ProfileTypes(ctx context.Context) ([]*ProfileType, error) { + ctx, span := tracing.DefaultTracer().Start(ctx, "datasource.pyroscope.ProfileTypes") + defer span.End() res, err := c.connectClient.ProfileTypes(ctx, connect.NewRequest(&querierv1.ProfileTypesRequest{})) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return nil, err } if res.Msg.ProfileTypes == nil { @@ -87,6 +95,8 @@ func (c *PyroscopeClient) ProfileTypes(ctx context.Context) ([]*ProfileType, err } func (c *PyroscopeClient) GetSeries(ctx context.Context, profileTypeID string, labelSelector string, start int64, end int64, groupBy []string, step float64) (*SeriesResponse, error) { + ctx, span := tracing.DefaultTracer().Start(ctx, "datasource.pyroscope.GetSeries", trace.WithAttributes(attribute.String("profileTypeID", profileTypeID), attribute.String("labelSelector", labelSelector))) + defer span.End() req := connect.NewRequest(&querierv1.SelectSeriesRequest{ ProfileTypeID: profileTypeID, LabelSelector: labelSelector, @@ -98,6 +108,8 @@ func (c *PyroscopeClient) GetSeries(ctx context.Context, profileTypeID string, l resp, err := c.connectClient.SelectSeries(ctx, req) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return nil, err } @@ -136,6 +148,8 @@ func (c *PyroscopeClient) GetSeries(ctx context.Context, profileTypeID string, l } func (c *PyroscopeClient) GetProfile(ctx context.Context, profileTypeID, labelSelector string, start, end int64, maxNodes *int64) (*ProfileResponse, error) { + ctx, span := tracing.DefaultTracer().Start(ctx, "datasource.pyroscope.GetProfile", trace.WithAttributes(attribute.String("profileTypeID", profileTypeID), attribute.String("labelSelector", labelSelector))) + defer span.End() req := &connect.Request[querierv1.SelectMergeStacktracesRequest]{ Msg: &querierv1.SelectMergeStacktracesRequest{ ProfileTypeID: profileTypeID, @@ -148,6 +162,8 @@ func (c *PyroscopeClient) GetProfile(ctx context.Context, profileTypeID, labelSe resp, err := c.connectClient.SelectMergeStacktraces(ctx, req) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return nil, err } @@ -187,8 +203,12 @@ func getUnits(profileTypeID string) string { } func (c *PyroscopeClient) LabelNames(ctx context.Context) ([]string, error) { + ctx, span := tracing.DefaultTracer().Start(ctx, "datasource.pyroscope.LabelNames") + defer span.End() resp, err := c.connectClient.LabelNames(ctx, connect.NewRequest(&typesv1.LabelNamesRequest{})) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return nil, fmt.Errorf("error seding LabelNames request %v", err) } @@ -203,8 +223,12 @@ func (c *PyroscopeClient) LabelNames(ctx context.Context) ([]string, error) { } func (c *PyroscopeClient) LabelValues(ctx context.Context, label string) ([]string, error) { + ctx, span := tracing.DefaultTracer().Start(ctx, "datasource.pyroscope.LabelValues") + defer span.End() resp, err := c.connectClient.LabelValues(ctx, connect.NewRequest(&typesv1.LabelValuesRequest{Name: label})) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return nil, err } return resp.Msg.Names, nil diff --git a/pkg/tsdb/grafana-pyroscope-datasource/query.go b/pkg/tsdb/grafana-pyroscope-datasource/query.go index 0bbd30a6e4c2..3f21c0cf8c8b 100644 --- a/pkg/tsdb/grafana-pyroscope-datasource/query.go +++ b/pkg/tsdb/grafana-pyroscope-datasource/query.go @@ -10,10 +10,14 @@ import ( "github.com/grafana/grafana-plugin-sdk-go/backend" "github.com/grafana/grafana-plugin-sdk-go/backend/gtime" + "github.com/grafana/grafana-plugin-sdk-go/backend/tracing" "github.com/grafana/grafana-plugin-sdk-go/data" "github.com/grafana/grafana-plugin-sdk-go/live" "github.com/grafana/grafana/pkg/tsdb/grafana-pyroscope-datasource/kinds/dataquery" "github.com/xlab/treeprint" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/trace" "golang.org/x/sync/errgroup" ) @@ -34,11 +38,16 @@ const ( // query processes single Pyroscope query transforming the response to data.Frame packaged in DataResponse func (d *PyroscopeDatasource) query(ctx context.Context, pCtx backend.PluginContext, query backend.DataQuery) backend.DataResponse { + ctx, span := tracing.DefaultTracer().Start(ctx, "datasource.pyroscope.query", trace.WithAttributes(attribute.String("query_type", query.QueryType))) + defer span.End() + var qm queryModel response := backend.DataResponse{} err := json.Unmarshal(query.JSON, &qm) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) response.Error = fmt.Errorf("error unmarshaling query model: %v", err) return response } @@ -50,6 +59,8 @@ func (d *PyroscopeDatasource) query(ctx context.Context, pCtx backend.PluginCont var dsJson dsJsonModel err = json.Unmarshal(pCtx.DataSourceInstanceSettings.JSONData, &dsJson) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return fmt.Errorf("error unmarshaling datasource json model: %v", err) } @@ -72,6 +83,8 @@ func (d *PyroscopeDatasource) query(ctx context.Context, pCtx backend.PluginCont math.Max(query.Interval.Seconds(), parsedInterval.Seconds()), ) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) logger.Error("Querying SelectSeries()", "err", err) return err } @@ -88,6 +101,8 @@ func (d *PyroscopeDatasource) query(ctx context.Context, pCtx backend.PluginCont logger.Debug("Calling GetProfile", "queryModel", qm) prof, err := d.client.GetProfile(gCtx, qm.ProfileTypeId, qm.LabelSelector, query.TimeRange.From.UnixMilli(), query.TimeRange.To.UnixMilli(), qm.MaxNodes) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) logger.Error("Error GetProfile()", "err", err) return err } @@ -120,6 +135,8 @@ func (d *PyroscopeDatasource) query(ctx context.Context, pCtx backend.PluginCont } if err := g.Wait(); err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) response.Error = g.Wait() } diff --git a/pkg/tsdb/parca/plugin.go b/pkg/tsdb/parca/plugin.go index fd46dbe6ace0..0ca260aebe97 100644 --- a/pkg/tsdb/parca/plugin.go +++ b/pkg/tsdb/parca/plugin.go @@ -8,7 +8,10 @@ import ( "github.com/bufbuild/connect-go" "github.com/grafana/grafana-plugin-sdk-go/backend" "github.com/grafana/grafana-plugin-sdk-go/backend/instancemgmt" + "github.com/grafana/grafana-plugin-sdk-go/backend/tracing" "github.com/grafana/grafana/pkg/infra/httpclient" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" ) // Make sure ParcaDatasource implements required interfaces. This is important to do @@ -56,6 +59,9 @@ func (d *ParcaDatasource) Dispose() { func (d *ParcaDatasource) CallResource(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error { logger.Debug("CallResource", "Path", req.Path, "Method", req.Method, "Body", req.Body) + ctx, span := tracing.DefaultTracer().Start(ctx, "datasource.parca.CallResource", trace.WithAttributes(attribute.String("path", req.Path), attribute.String("method", req.Method))) + defer span.End() + if req.Path == "profileTypes" { return d.callProfileTypes(ctx, req, sender) } diff --git a/pkg/tsdb/parca/query.go b/pkg/tsdb/parca/query.go index 07618f8a521a..83ee24b67697 100644 --- a/pkg/tsdb/parca/query.go +++ b/pkg/tsdb/parca/query.go @@ -10,8 +10,12 @@ import ( v1alpha1 "buf.build/gen/go/parca-dev/parca/protocolbuffers/go/parca/query/v1alpha1" "github.com/bufbuild/connect-go" "github.com/grafana/grafana-plugin-sdk-go/backend" + "github.com/grafana/grafana-plugin-sdk-go/backend/tracing" "github.com/grafana/grafana-plugin-sdk-go/data" "github.com/grafana/grafana/pkg/tsdb/parca/kinds/dataquery" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/trace" "google.golang.org/protobuf/types/known/timestamppb" ) @@ -27,12 +31,17 @@ const ( // query processes single Parca query transforming the response to data.Frame packaged in DataResponse func (d *ParcaDatasource) query(ctx context.Context, pCtx backend.PluginContext, query backend.DataQuery) backend.DataResponse { + ctx, span := tracing.DefaultTracer().Start(ctx, "datasource.parca.query", trace.WithAttributes(attribute.String("query_type", query.QueryType))) + defer span.End() + var qm queryModel response := backend.DataResponse{} err := json.Unmarshal(query.JSON, &qm) if err != nil { response.Error = err + span.RecordError(response.Error) + span.SetStatus(codes.Error, response.Error.Error()) return response } @@ -40,6 +49,8 @@ func (d *ParcaDatasource) query(ctx context.Context, pCtx backend.PluginContext, seriesResp, err := d.client.QueryRange(ctx, makeMetricRequest(qm, query)) if err != nil { response.Error = err + span.RecordError(response.Error) + span.SetStatus(codes.Error, response.Error.Error()) return response } response.Frames = append(response.Frames, seriesToDataFrame(seriesResp, qm.ProfileTypeId)...) @@ -50,6 +61,8 @@ func (d *ParcaDatasource) query(ctx context.Context, pCtx backend.PluginContext, resp, err := d.client.Query(ctx, makeProfileRequest(qm, query)) if err != nil { response.Error = err + span.RecordError(response.Error) + span.SetStatus(codes.Error, response.Error.Error()) return response } frame := responseToDataFrames(resp) diff --git a/pkg/tsdb/parca/resources.go b/pkg/tsdb/parca/resources.go index 87fad90a0e59..db27a9bfad43 100644 --- a/pkg/tsdb/parca/resources.go +++ b/pkg/tsdb/parca/resources.go @@ -9,6 +9,8 @@ import ( v1alpha1 "buf.build/gen/go/parca-dev/parca/protocolbuffers/go/parca/query/v1alpha1" "github.com/bufbuild/connect-go" "github.com/grafana/grafana-plugin-sdk-go/backend" + "github.com/grafana/grafana-plugin-sdk-go/backend/tracing" + "go.opentelemetry.io/otel/codes" ) type ProfileType struct { @@ -23,8 +25,12 @@ type ProfileType struct { } func (d *ParcaDatasource) callProfileTypes(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error { + ctx, span := tracing.DefaultTracer().Start(ctx, "datasource.parca.callProfileTypes") + defer span.End() res, err := d.client.ProfileTypes(ctx, connect.NewRequest(&v1alpha1.ProfileTypesRequest{})) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return err } @@ -50,35 +56,51 @@ func (d *ParcaDatasource) callProfileTypes(ctx context.Context, req *backend.Cal data, err := json.Marshal(types) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return err } err = sender.Send(&backend.CallResourceResponse{Body: data, Headers: req.Headers, Status: 200}) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return err } return nil } func (d *ParcaDatasource) callLabelNames(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error { + ctx, span := tracing.DefaultTracer().Start(ctx, "datasource.parca.callLabelNames") + defer span.End() res, err := d.client.Labels(ctx, connect.NewRequest(&v1alpha1.LabelsRequest{})) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return err } data, err := json.Marshal(res.Msg.LabelNames) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return err } err = sender.Send(&backend.CallResourceResponse{Body: data, Headers: req.Headers, Status: 200}) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return err } return nil } func (d *ParcaDatasource) callLabelValues(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error { + ctx, span := tracing.DefaultTracer().Start(ctx, "datasource.parca.callLabelValues") + defer span.End() parsedUrl, err := url.Parse(req.URL) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return err } label, ok := parsedUrl.Query()["label"] @@ -87,14 +109,20 @@ func (d *ParcaDatasource) callLabelValues(ctx context.Context, req *backend.Call } res, err := d.client.Values(ctx, connect.NewRequest(&v1alpha1.ValuesRequest{LabelName: label[0]})) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return err } data, err := json.Marshal(res.Msg.LabelValues) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return err } err = sender.Send(&backend.CallResourceResponse{Body: data, Headers: req.Headers, Status: 200}) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return err } return nil diff --git a/pkg/tsdb/tempo/search_stream.go b/pkg/tsdb/tempo/search_stream.go index bec6b987048b..d2ef5a1d8ae2 100644 --- a/pkg/tsdb/tempo/search_stream.go +++ b/pkg/tsdb/tempo/search_stream.go @@ -8,9 +8,12 @@ import ( "io" "github.com/grafana/grafana-plugin-sdk-go/backend" + "github.com/grafana/grafana-plugin-sdk-go/backend/tracing" "github.com/grafana/grafana-plugin-sdk-go/data" "github.com/grafana/grafana/pkg/tsdb/tempo/kinds/dataquery" "github.com/grafana/tempo/pkg/tempopb" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" ) const SearchPathPrefix = "search/" @@ -27,12 +30,17 @@ type StreamSender interface { } func (s *Service) runSearchStream(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender, datasource *Datasource) error { + ctx, span := tracing.DefaultTracer().Start(ctx, "datasource.tempo.runSearchStream") + defer span.End() + response := &backend.DataResponse{} var backendQuery *backend.DataQuery err := json.Unmarshal(req.Data, &backendQuery) if err != nil { response.Error = fmt.Errorf("error unmarshaling backend query model: %v", err) + span.RecordError(response.Error) + span.SetStatus(codes.Error, response.Error.Error()) return err } @@ -40,6 +48,8 @@ func (s *Service) runSearchStream(ctx context.Context, req *backend.RunStreamReq err = json.Unmarshal(req.Data, &sr) if err != nil { response.Error = fmt.Errorf("error unmarshaling Tempo query model: %v", err) + span.RecordError(response.Error) + span.SetStatus(codes.Error, response.Error.Error()) return err } @@ -52,46 +62,60 @@ func (s *Service) runSearchStream(ctx context.Context, req *backend.RunStreamReq stream, err := datasource.StreamingClient.Search(ctx, sr) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) s.logger.Error("Error Search()", "err", err) return err } - return s.processStream(stream, sender) + return s.processStream(ctx, stream, sender) } -func (s *Service) processStream(stream tempopb.StreamingQuerier_SearchClient, sender StreamSender) error { +func (s *Service) processStream(ctx context.Context, stream tempopb.StreamingQuerier_SearchClient, sender StreamSender) error { + ctx, span := tracing.DefaultTracer().Start(ctx, "datasource.tempo.processStream") + defer span.End() var traceList []*tempopb.TraceSearchMetadata var metrics *tempopb.SearchMetrics + messageCount := 0 for { msg, err := stream.Recv() + messageCount++ + span.SetAttributes(attribute.Int("message_count", messageCount)) if errors.Is(err, io.EOF) { - if err := sendResponse(&ExtendedResponse{ + if err := s.sendResponse(ctx, &ExtendedResponse{ State: dataquery.SearchStreamingStateDone, SearchResponse: &tempopb.SearchResponse{ Metrics: metrics, Traces: traceList, }, }, sender); err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return err } break } if err != nil { s.logger.Error("Error receiving message", "err", err) + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return err } metrics = msg.Metrics traceList = append(traceList, msg.Traces...) traceList = removeDuplicates(traceList) + span.SetAttributes(attribute.Int("traces_count", len(traceList))) - if err := sendResponse(&ExtendedResponse{ + if err := s.sendResponse(ctx, &ExtendedResponse{ State: dataquery.SearchStreamingStateStreaming, SearchResponse: &tempopb.SearchResponse{ Metrics: metrics, Traces: traceList, }, }, sender); err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return err } } @@ -99,10 +123,14 @@ func (s *Service) processStream(stream tempopb.StreamingQuerier_SearchClient, se return nil } -func sendResponse(response *ExtendedResponse, sender StreamSender) error { +func (s *Service) sendResponse(ctx context.Context, response *ExtendedResponse, sender StreamSender) error { + _, span := tracing.DefaultTracer().Start(ctx, "datasource.tempo.sendResponse") + defer span.End() frame := createResponseDataFrame() if response != nil { + span.SetAttributes(attribute.Int("trace_count", len(response.Traces)), attribute.String("state", string(response.State))) + tracesAsJson, err := json.Marshal(response.Traces) if err != nil { return err diff --git a/pkg/tsdb/tempo/search_stream_test.go b/pkg/tsdb/tempo/search_stream_test.go index 035d46da9c55..dddda21d2012 100644 --- a/pkg/tsdb/tempo/search_stream_test.go +++ b/pkg/tsdb/tempo/search_stream_test.go @@ -21,7 +21,7 @@ func TestProcessStream_ValidInput_ReturnsNoError(t *testing.T) { service := &Service{} searchClient := &mockStreamer{} streamSender := &mockSender{} - err := service.processStream(searchClient, streamSender) + err := service.processStream(context.Background(), searchClient, streamSender) if err != nil { t.Errorf("Expected no error, but got %s", err) } @@ -33,7 +33,7 @@ func TestProcessStream_InvalidInput_ReturnsError(t *testing.T) { } searchClient := &mockStreamer{err: errors.New("invalid input")} streamSender := &mockSender{} - err := service.processStream(searchClient, streamSender) + err := service.processStream(context.Background(), searchClient, streamSender) if err != nil { if !strings.Contains(err.Error(), "invalid input") { t.Errorf("Expected error message to contain 'invalid input', but got %s", err) @@ -109,7 +109,7 @@ func TestProcessStream_ValidInput_ReturnsExpectedOutput(t *testing.T) { }, } streamSender := &mockSender{} - err := service.processStream(searchClient, streamSender) + err := service.processStream(context.Background(), searchClient, streamSender) if err != nil { t.Errorf("Expected no error, but got %s", err) return diff --git a/pkg/tsdb/tempo/trace.go b/pkg/tsdb/tempo/trace.go index 9bd4ff33f6ef..8b3ac71aef78 100644 --- a/pkg/tsdb/tempo/trace.go +++ b/pkg/tsdb/tempo/trace.go @@ -8,15 +8,24 @@ import ( "net/http" "github.com/grafana/grafana-plugin-sdk-go/backend" + "github.com/grafana/grafana-plugin-sdk-go/backend/tracing" "github.com/grafana/grafana-plugin-sdk-go/data" "github.com/grafana/grafana/pkg/tsdb/tempo/kinds/dataquery" "go.opentelemetry.io/collector/pdata/ptrace" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/trace" ) func (s *Service) getTrace(ctx context.Context, pCtx backend.PluginContext, query backend.DataQuery) (*backend.DataResponse, error) { result := &backend.DataResponse{} refID := query.RefID + ctx, span := tracing.DefaultTracer().Start(ctx, "datasource.tempo.getTrace", trace.WithAttributes( + attribute.String("queryType", query.QueryType), + )) + defer span.End() + model := &dataquery.TempoQuery{} err := json.Unmarshal(query.JSON, model) if err != nil { @@ -34,11 +43,15 @@ func (s *Service) getTrace(ctx context.Context, pCtx backend.PluginContext, quer request, err := s.createRequest(ctx, dsInfo, *model.Query, query.TimeRange.From.Unix(), query.TimeRange.To.Unix()) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return result, err } resp, err := dsInfo.HTTPClient.Do(request) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return result, fmt.Errorf("failed get to tempo: %w", err) } @@ -55,6 +68,8 @@ func (s *Service) getTrace(ctx context.Context, pCtx backend.PluginContext, quer if resp.StatusCode != http.StatusOK { result.Error = fmt.Errorf("failed to get trace with id: %v Status: %s Body: %s", model.Query, resp.Status, string(body)) + span.RecordError(result.Error) + span.SetStatus(codes.Error, result.Error.Error()) return result, nil } @@ -62,11 +77,15 @@ func (s *Service) getTrace(ctx context.Context, pCtx backend.PluginContext, quer otTrace, err := pbUnmarshaler.UnmarshalTraces(body) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return &backend.DataResponse{}, fmt.Errorf("failed to convert tempo response to Otlp: %w", err) } frame, err := TraceToFrame(otTrace) if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, err.Error()) return &backend.DataResponse{}, fmt.Errorf("failed to transform trace %v to data frame: %w", model.Query, err) } frame.RefID = refID