mirror of
https://github.com/grafana/grafana.git
synced 2024-11-29 20:24:18 -06:00
72118e372c
* Remove unneeded log * Add elapsed time to streaming progress table * Use disableTraceQLStreaming toggle to determine default status of streaming
100 lines
2.5 KiB
Go
100 lines
2.5 KiB
Go
package tempo
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"net/http"
|
|
|
|
"github.com/grafana/grafana-plugin-sdk-go/backend"
|
|
"github.com/grafana/grafana-plugin-sdk-go/backend/datasource"
|
|
"github.com/grafana/grafana-plugin-sdk-go/backend/instancemgmt"
|
|
"github.com/grafana/grafana/pkg/infra/httpclient"
|
|
"github.com/grafana/grafana/pkg/infra/log"
|
|
"github.com/grafana/grafana/pkg/tsdb/tempo/kinds/dataquery"
|
|
"github.com/grafana/tempo/pkg/tempopb"
|
|
)
|
|
|
|
type Service struct {
|
|
im instancemgmt.InstanceManager
|
|
logger log.Logger
|
|
}
|
|
|
|
func ProvideService(httpClientProvider httpclient.Provider) *Service {
|
|
return &Service{
|
|
logger: log.New("tsdb.tempo"),
|
|
im: datasource.NewInstanceManager(newInstanceSettings(httpClientProvider)),
|
|
}
|
|
}
|
|
|
|
type Datasource struct {
|
|
HTTPClient *http.Client
|
|
StreamingClient tempopb.StreamingQuerierClient
|
|
URL string
|
|
}
|
|
|
|
func newInstanceSettings(httpClientProvider httpclient.Provider) datasource.InstanceFactoryFunc {
|
|
return func(settings backend.DataSourceInstanceSettings) (instancemgmt.Instance, error) {
|
|
opts, err := settings.HTTPClientOptions()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
client, err := httpClientProvider.New(opts)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
streamingClient, err := newGrpcClient(settings, opts)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
model := &Datasource{
|
|
HTTPClient: client,
|
|
StreamingClient: streamingClient,
|
|
URL: settings.URL,
|
|
}
|
|
return model, nil
|
|
}
|
|
}
|
|
|
|
func (s *Service) QueryData(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
|
|
// create response struct
|
|
response := backend.NewQueryDataResponse()
|
|
|
|
// loop over queries and execute them individually.
|
|
for _, q := range req.Queries {
|
|
if res, err := s.query(ctx, req.PluginContext, q); err != nil {
|
|
return response, err
|
|
} else {
|
|
if res != nil {
|
|
response.Responses[q.RefID] = *res
|
|
}
|
|
}
|
|
}
|
|
|
|
return response, nil
|
|
}
|
|
|
|
func (s *Service) query(ctx context.Context, pCtx backend.PluginContext, query backend.DataQuery) (*backend.DataResponse, error) {
|
|
if query.QueryType == string(dataquery.TempoQueryTypeTraceId) {
|
|
return s.getTrace(ctx, pCtx, query)
|
|
}
|
|
|
|
return nil, fmt.Errorf("unsupported query type: '%s' for query with refID '%s'", query.QueryType, query.RefID)
|
|
}
|
|
|
|
func (s *Service) getDSInfo(ctx context.Context, pluginCtx backend.PluginContext) (*Datasource, error) {
|
|
i, err := s.im.Get(ctx, pluginCtx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
instance, ok := i.(*Datasource)
|
|
if !ok {
|
|
return nil, fmt.Errorf("failed to cast datsource info")
|
|
}
|
|
|
|
return instance, nil
|
|
}
|