Query Service: Combine SSE handling in single tenant and multi tenant paths (#108041)
* parse via sse I need to figure out how to handle the pipeline.execute with our own client. I think this is important for MT reasons, just like using our own cache (via legacy) is important. parsing is done though! * WIP nonsense * horrible code but i think it works * Add support for sql expressions config settings * Cleanup: - remove spew from nodes.go - uncomment out plugin context and use in single tenant flow - make code more readable and add comments * Cleanup: - create separate file for mt ds client builder - ensure error handling is the same for both expressions and regular queries - other cleanup * not working but good thoughts * WIP, vector not working for non sse * super hacky but i think vectors work now * delete delete delete * Comments for future ref * break out query handling and start test * add prom debugger * clean up: remove comments and commented out bits * fix query_test * add prom debugger * create table-driven tests with testsdata files * Fix test * Add test * go mod?? * idk * Remove comment * go enterprise issue maybe * Fix codeowners * Delete * Remove test data * Clean up * logger * Remove go changes hopefully * idk go man * sad * idk i ran go mod tidy and this is what it wants * Fix readme, with much help from adam * some linting and testing errors * lint * fix lint * fix lint register.go * another lint * address lint in test * fix dead code and linters for query_test * Go mod? * Struggling with go mod * Fix test * Fix another test * Revert headers change * Its difficult to test this in OSS as it depends on functionality defined in enterprise, let's bring these tests back in some form in enterprise * Fix codeowners --------- Co-authored-by: Adam Simpson <adam@adamsimpson.net>
This commit is contained in:
Sarah Zinger
co-authored by
Adam Simpson
Adam Simpson
parent
8acdc7e37b
commit
3fad863fd1
@@ -2,6 +2,7 @@ package query
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
@@ -13,6 +14,7 @@ import (
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend/gtime"
|
||||
"golang.org/x/sync/errgroup"
|
||||
|
||||
data "github.com/grafana/grafana-plugin-sdk-go/experimental/apis/data/v0alpha1"
|
||||
"github.com/grafana/grafana/pkg/api/dtos"
|
||||
"github.com/grafana/grafana/pkg/apimachinery/errutil"
|
||||
"github.com/grafana/grafana/pkg/apimachinery/identity"
|
||||
@@ -22,6 +24,7 @@ import (
|
||||
"github.com/grafana/grafana/pkg/plugins"
|
||||
"github.com/grafana/grafana/pkg/services/contexthandler"
|
||||
"github.com/grafana/grafana/pkg/services/datasources"
|
||||
"github.com/grafana/grafana/pkg/services/mtdsclient"
|
||||
"github.com/grafana/grafana/pkg/services/pluginsintegration/plugincontext"
|
||||
"github.com/grafana/grafana/pkg/services/validations"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
@@ -47,6 +50,7 @@ func ProvideService(
|
||||
dataSourceRequestValidator validations.DataSourceRequestValidator,
|
||||
pluginClient plugins.Client,
|
||||
pCtxProvider *plugincontext.Provider,
|
||||
mtDatasourceClientBuilder mtdsclient.MTDatasourceClientBuilder,
|
||||
) *ServiceImpl {
|
||||
g := &ServiceImpl{
|
||||
cfg: cfg,
|
||||
@@ -57,6 +61,7 @@ func ProvideService(
|
||||
pCtxProvider: pCtxProvider,
|
||||
log: log.New("query_data"),
|
||||
concurrentQueryLimit: cfg.SectionWithEnvOverrides("query").Key("concurrent_query_limit").MustInt(runtime.NumCPU()),
|
||||
mtDatasourceClientBuilder: mtDatasourceClientBuilder,
|
||||
}
|
||||
g.log.Info("Query Service initialization")
|
||||
return g
|
||||
@@ -80,6 +85,8 @@ type ServiceImpl struct {
|
||||
pCtxProvider *plugincontext.Provider
|
||||
log log.Logger
|
||||
concurrentQueryLimit int
|
||||
mtDatasourceClientBuilder mtdsclient.MTDatasourceClientBuilder
|
||||
headers map[string]string
|
||||
}
|
||||
|
||||
// Run ServiceImpl.
|
||||
@@ -201,7 +208,19 @@ func buildErrorResponses(err error, queries []*simplejson.Json) splitResponse {
|
||||
return splitResponse{er, http.Header{}}
|
||||
}
|
||||
|
||||
// handleExpressions handles POST /api/ds/query when there is an expression.
|
||||
func QueryData(ctx context.Context, log log.Logger, dscache datasources.CacheService, exprService *expr.Service, reqDTO dtos.MetricRequest, mtDatasourceClientBuilder mtdsclient.MTDatasourceClientBuilder, headers map[string]string) (*backend.QueryDataResponse, error) {
|
||||
s := &ServiceImpl{
|
||||
log: log,
|
||||
dataSourceCache: dscache,
|
||||
expressionService: exprService,
|
||||
dataSourceRequestValidator: validations.ProvideValidator(),
|
||||
mtDatasourceClientBuilder: mtDatasourceClientBuilder,
|
||||
headers: headers,
|
||||
}
|
||||
return s.QueryData(ctx, nil, false, reqDTO)
|
||||
}
|
||||
|
||||
// handleExpressions handles queries when there is an expression.
|
||||
func (s *ServiceImpl) handleExpressions(ctx context.Context, user identity.Requester, parsedReq *parsedRequest) (*backend.QueryDataResponse, error) {
|
||||
exprReq := expr.Request{
|
||||
Queries: []expr.Query{},
|
||||
@@ -257,21 +276,37 @@ func (s *ServiceImpl) handleQuerySingleDatasource(ctx context.Context, user iden
|
||||
}
|
||||
}
|
||||
|
||||
pCtx, err := s.pCtxProvider.GetWithDataSource(ctx, ds.Type, user, ds)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
req := &backend.QueryDataRequest{
|
||||
PluginContext: pCtx,
|
||||
Headers: map[string]string{},
|
||||
Queries: []backend.DataQuery{},
|
||||
Headers: map[string]string{},
|
||||
Queries: []backend.DataQuery{},
|
||||
}
|
||||
|
||||
for _, q := range queries {
|
||||
req.Queries = append(req.Queries, q.query)
|
||||
}
|
||||
|
||||
return s.pluginClient.QueryData(ctx, req)
|
||||
mtDsClient, ok := s.mtDatasourceClientBuilder.BuildClient(ds.Type, ds.UID)
|
||||
if !ok { // single tenant flow
|
||||
pCtx, err := s.pCtxProvider.GetWithDataSource(ctx, ds.Type, user, ds)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
req.PluginContext = pCtx
|
||||
return s.pluginClient.QueryData(ctx, req)
|
||||
} else { // multi tenant flow
|
||||
// transform request from backend.QueryDataRequest to k8s request
|
||||
k8sReq := &data.QueryDataRequest{}
|
||||
for _, q := range req.Queries {
|
||||
var dataQuery data.DataQuery
|
||||
err := json.Unmarshal(q.JSON, &dataQuery)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
k8sReq.Queries = append(k8sReq.Queries, dataQuery)
|
||||
}
|
||||
return mtDsClient.QueryData(ctx, *k8sReq)
|
||||
}
|
||||
}
|
||||
|
||||
// parseRequest parses a request into parsed queries grouped by datasource uid
|
||||
|
||||
@@ -30,6 +30,7 @@ import (
|
||||
"github.com/grafana/grafana/pkg/services/datasources"
|
||||
fakeDatasources "github.com/grafana/grafana/pkg/services/datasources/fakes"
|
||||
"github.com/grafana/grafana/pkg/services/featuremgmt"
|
||||
"github.com/grafana/grafana/pkg/services/mtdsclient"
|
||||
"github.com/grafana/grafana/pkg/services/pluginsintegration/pluginconfig"
|
||||
"github.com/grafana/grafana/pkg/services/pluginsintegration/plugincontext"
|
||||
pluginSettings "github.com/grafana/grafana/pkg/services/pluginsintegration/pluginsettings/service"
|
||||
@@ -483,8 +484,8 @@ func setup(t *testing.T) *testContext {
|
||||
pluginSettings.ProvideService(sqlStore, secretsService), pluginconfig.NewFakePluginRequestConfigProvider(),
|
||||
)
|
||||
exprService := expr.ProvideService(&setting.Cfg{ExpressionsEnabled: true}, pc, pCtxProvider,
|
||||
featuremgmt.WithFeatures(), nil, tracing.InitializeTracerForTest())
|
||||
queryService := ProvideService(setting.NewCfg(), dc, exprService, rv, pc, pCtxProvider) // provider belonging to this package
|
||||
featuremgmt.WithFeatures(), nil, tracing.InitializeTracerForTest(), mtdsclient.NewNullMTDatasourceClientBuilder())
|
||||
queryService := ProvideService(setting.NewCfg(), dc, exprService, rv, pc, pCtxProvider, mtdsclient.NewNullMTDatasourceClientBuilder()) // provider belonging to this package
|
||||
return &testContext{
|
||||
pluginContext: pc,
|
||||
secretStore: ss,
|
||||
|
||||
Reference in New Issue
Block a user