mirror of
https://github.com/grafana/grafana.git
synced 2024-11-28 19:54:10 -06:00
c1709c9301
* Refactor Tempo datasource backend to support multiple queryData types. Added traceId query type that is set when performing the request but doesn't map to a tab. * WIP data is reaching the frontend * WIP * Use channels and goroutines * Some fixes * Simplify backend code. Return traces, metrics, state and error in a dataframe. Shared state type between FE and BE. Use getStream() instead of getQueryData() * Handle errors in frontend * Update Tempo and use same URL for RPC and HTTP * Cleanup backend code * Merge main * Create grpc client only with host and authenticate * Create grpc client only with host and authenticate * Cleanup * Add streaming to TraceQL Search tab * Fix merge conflicts * Added tests for processStream * make gen-cue * make gen-cue * goimports * lint * Cleanup go.mod * Comments * Addressing PR comments * Fix streaming for tracel search tab * Added streaming kill switch as the disableTraceQLStreaming feature toggle * Small comment * Fix conflicts * Correctly capture and send all errors as a DF to client * Fix infinite error loop * Fix merge conflicts * Fix test * Update deprecated import * Fix feature toggles gen * Fix merge conflicts
49 lines
1.4 KiB
Go
49 lines
1.4 KiB
Go
package tempo
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strings"
|
|
|
|
"github.com/grafana/grafana-plugin-sdk-go/backend"
|
|
)
|
|
|
|
func (s *Service) SubscribeStream(ctx context.Context, req *backend.SubscribeStreamRequest) (*backend.SubscribeStreamResponse, error) {
|
|
s.logger.Debug("Allowing access to stream", "path", req.Path, "user", req.PluginContext.User)
|
|
status := backend.SubscribeStreamStatusPermissionDenied
|
|
if strings.HasPrefix(req.Path, SearchPathPrefix) {
|
|
status = backend.SubscribeStreamStatusOK
|
|
}
|
|
|
|
return &backend.SubscribeStreamResponse{
|
|
Status: status,
|
|
}, nil
|
|
}
|
|
|
|
func (s *Service) PublishStream(ctx context.Context, req *backend.PublishStreamRequest) (*backend.PublishStreamResponse, error) {
|
|
s.logger.Debug("PublishStream called")
|
|
|
|
// Do not allow publishing at all.
|
|
return &backend.PublishStreamResponse{
|
|
Status: backend.PublishStreamStatusPermissionDenied,
|
|
}, nil
|
|
}
|
|
|
|
func (s *Service) RunStream(ctx context.Context, request *backend.RunStreamRequest, sender *backend.StreamSender) error {
|
|
s.logger.Debug("New stream call", "path", request.Path)
|
|
|
|
if strings.HasPrefix(request.Path, SearchPathPrefix) {
|
|
tempoDatasource, err := s.getDSInfo(ctx, request.PluginContext)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err = s.runSearchStream(ctx, request, sender, tempoDatasource); err != nil {
|
|
return sendError(err, sender)
|
|
} else {
|
|
return nil
|
|
}
|
|
}
|
|
|
|
return fmt.Errorf("unknown path %s", request.Path)
|
|
}
|