public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Plugins: Enhanced plugin instrumentation (#90199)

Browse Source 
* Plugins: Enhanced plugin instrumentation

* use backend.CallResourceResponseSenderFunc

* sdk v0.237.0

* support admission control

* cover all handlers in log and metrics middlewares

* fix after review
This commit is contained in:
Marcus Efraimsson
2024-07-09 15:03:46 +02:00
committed by GitHub
parent 2a4a73e03d
commit c8af659f02
40 changed files with 700 additions and 472 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/services/pluginsintegration/clientmiddleware/caching_middleware.go
+1 -1
View File
@@ -160,7 +160,7 @@ func (m *CachingMiddleware) CallResource(ctx context.Context, req *backend.CallR
return m.next.CallResource(ctx, req, sender)
}
// Otherwise, intercept the responses in a wrapped sender so we can cache them first
cacheSender := callResourceResponseSenderFunc(func(res *backend.CallResourceResponse) error {
cacheSender := backend.CallResourceResponseSenderFunc(func(res *backend.CallResourceResponse) error {
cr.UpdateCacheFn(ctx, res)
return sender.Send(res)
})
pkg/services/pluginsintegration/clientmiddleware/caching_middleware_test.go
+1 -1
View File
@@ -221,7 +221,7 @@ func TestCachingMiddleware(t *testing.T) {
}
var sentResponse *backend.CallResourceResponse
var storeOneResponseCallResourceSender = callResourceResponseSenderFunc(func(res *backend.CallResourceResponse) error {
var storeOneResponseCallResourceSender = backend.CallResourceResponseSenderFunc(func(res *backend.CallResourceResponse) error {
sentResponse = res
return nil
})
pkg/services/pluginsintegration/clientmiddleware/contextual_logger_middleware.go
+19 -12
View File
@@ -24,8 +24,15 @@ type ContextualLoggerMiddleware struct {
}
// instrumentContext adds a contextual logger with plugin and request details to the given context.
func instrumentContext(ctx context.Context, endpoint string, pCtx backend.PluginContext) context.Context {
p := []any{"endpoint", endpoint, "pluginId", pCtx.PluginID}
func instrumentContext(ctx context.Context, pCtx backend.PluginContext) context.Context {
p := []any{}
if ep := backend.EndpointFromContext(ctx); !ep.IsEmpty() {
p = append(p, "endpoint", string(ep))
}
p = append(p, "pluginId", pCtx.PluginID)
if pCtx.DataSourceInstanceSettings != nil {
p = append(p, "dsName", pCtx.DataSourceInstanceSettings.Name)
p = append(p, "dsUID", pCtx.DataSourceInstanceSettings.UID)
@@ -37,54 +44,54 @@ func instrumentContext(ctx context.Context, endpoint string, pCtx backend.Plugin
}
func (m *ContextualLoggerMiddleware) QueryData(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
ctx = instrumentContext(ctx, endpointQueryData, req.PluginContext)
ctx = instrumentContext(ctx, req.PluginContext)
return m.next.QueryData(ctx, req)
}
func (m *ContextualLoggerMiddleware) CallResource(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error {
ctx = instrumentContext(ctx, endpointCallResource, req.PluginContext)
ctx = instrumentContext(ctx, req.PluginContext)
return m.next.CallResource(ctx, req, sender)
}
func (m *ContextualLoggerMiddleware) CheckHealth(ctx context.Context, req *backend.CheckHealthRequest) (*backend.CheckHealthResult, error) {
ctx = instrumentContext(ctx, endpointCheckHealth, req.PluginContext)
ctx = instrumentContext(ctx, req.PluginContext)
return m.next.CheckHealth(ctx, req)
}
func (m *ContextualLoggerMiddleware) CollectMetrics(ctx context.Context, req *backend.CollectMetricsRequest) (*backend.CollectMetricsResult, error) {
ctx = instrumentContext(ctx, endpointCollectMetrics, req.PluginContext)
ctx = instrumentContext(ctx, req.PluginContext)
return m.next.CollectMetrics(ctx, req)
}
func (m *ContextualLoggerMiddleware) SubscribeStream(ctx context.Context, req *backend.SubscribeStreamRequest) (*backend.SubscribeStreamResponse, error) {
ctx = instrumentContext(ctx, endpointSubscribeStream, req.PluginContext)
ctx = instrumentContext(ctx, req.PluginContext)
return m.next.SubscribeStream(ctx, req)
}
func (m *ContextualLoggerMiddleware) PublishStream(ctx context.Context, req *backend.PublishStreamRequest) (*backend.PublishStreamResponse, error) {
ctx = instrumentContext(ctx, endpointPublishStream, req.PluginContext)
ctx = instrumentContext(ctx, req.PluginContext)
return m.next.PublishStream(ctx, req)
}
func (m *ContextualLoggerMiddleware) RunStream(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
ctx = instrumentContext(ctx, endpointRunStream, req.PluginContext)
ctx = instrumentContext(ctx, req.PluginContext)
return m.next.RunStream(ctx, req, sender)
}
// ValidateAdmission implements backend.AdmissionHandler.
func (m *ContextualLoggerMiddleware) ValidateAdmission(ctx context.Context, req *backend.AdmissionRequest) (*backend.ValidationResponse, error) {
ctx = instrumentContext(ctx, endpointValidateAdmission, req.PluginContext)
ctx = instrumentContext(ctx, req.PluginContext)
return m.next.ValidateAdmission(ctx, req)
}
// MutateAdmission implements backend.AdmissionHandler.
func (m *ContextualLoggerMiddleware) MutateAdmission(ctx context.Context, req *backend.AdmissionRequest) (*backend.MutationResponse, error) {
ctx = instrumentContext(ctx, endpointMutateAdmission, req.PluginContext)
ctx = instrumentContext(ctx, req.PluginContext)
return m.next.MutateAdmission(ctx, req)
}
// ConvertObject implements backend.AdmissionHandler.
func (m *ContextualLoggerMiddleware) ConvertObject(ctx context.Context, req *backend.ConversionRequest) (*backend.ConversionResponse, error) {
ctx = instrumentContext(ctx, endpointConvertObject, req.PluginContext)
ctx = instrumentContext(ctx, req.PluginContext)
return m.next.ConvertObject(ctx, req)
}
pkg/services/pluginsintegration/clientmiddleware/logger_middleware.go
+105 -13
View File
@@ -8,6 +8,7 @@ import (
"github.com/grafana/grafana/pkg/infra/log"
"github.com/grafana/grafana/pkg/plugins"
"github.com/grafana/grafana/pkg/plugins/instrumentationutils"
plog "github.com/grafana/grafana/pkg/plugins/log"
"github.com/grafana/grafana/pkg/plugins/pluginrequestmeta"
)
@@ -30,13 +31,13 @@ type LoggerMiddleware struct {
logger plog.Logger
}
func (m *LoggerMiddleware) logRequest(ctx context.Context, fn func(ctx context.Context) (requestStatus, error)) error {
func (m *LoggerMiddleware) logRequest(ctx context.Context, fn func(ctx context.Context) (instrumentationutils.RequestStatus, error)) error {
start := time.Now()
timeBeforePluginRequest := log.TimeSinceStart(ctx, start)
status, err := fn(ctx)
logParams := []any{
"status", status,
"status", status.String(),
"duration", time.Since(start),
"eventName", "grafana-data-egress",
"time_before_plugin_request", timeBeforePluginRequest,
@@ -48,7 +49,7 @@ func (m *LoggerMiddleware) logRequest(ctx context.Context, fn func(ctx context.C
ctxLogger := m.logger.FromContext(ctx)
logFunc := ctxLogger.Info
if status > requestStatusOK {
if status > instrumentationutils.RequestStatusOK {
logFunc = ctxLogger.Error
}
@@ -63,11 +64,12 @@ func (m *LoggerMiddleware) QueryData(ctx context.Context, req *backend.QueryData
}
var resp *backend.QueryDataResponse
err := m.logRequest(ctx, func(ctx context.Context) (status requestStatus, innerErr error) {
err := m.logRequest(ctx, func(ctx context.Context) (instrumentationutils.RequestStatus, error) {
var innerErr error
resp, innerErr = m.next.QueryData(ctx, req)
if innerErr != nil {
return requestStatusFromError(innerErr), innerErr
return instrumentationutils.RequestStatusFromError(innerErr), innerErr
}
ctxLogger := m.logger.FromContext(ctx)
@@ -83,7 +85,7 @@ func (m *LoggerMiddleware) QueryData(ctx context.Context, req *backend.QueryData
}
}
return requestStatusFromQueryDataResponse(resp, innerErr), innerErr
return instrumentationutils.RequestStatusFromQueryDataResponse(resp, innerErr), innerErr
})
return resp, err
@@ -94,9 +96,9 @@ func (m *LoggerMiddleware) CallResource(ctx context.Context, req *backend.CallRe
return m.next.CallResource(ctx, req, sender)
}
err := m.logRequest(ctx, func(ctx context.Context) (status requestStatus, innerErr error) {
innerErr = m.next.CallResource(ctx, req, sender)
return requestStatusFromError(innerErr), innerErr
err := m.logRequest(ctx, func(ctx context.Context) (instrumentationutils.RequestStatus, error) {
innerErr := m.next.CallResource(ctx, req, sender)
return instrumentationutils.RequestStatusFromError(innerErr), innerErr
})
return err
@@ -108,9 +110,10 @@ func (m *LoggerMiddleware) CheckHealth(ctx context.Context, req *backend.CheckHe
}
var resp *backend.CheckHealthResult
err := m.logRequest(ctx, func(ctx context.Context) (status requestStatus, innerErr error) {
err := m.logRequest(ctx, func(ctx context.Context) (instrumentationutils.RequestStatus, error) {
var innerErr error
resp, innerErr = m.next.CheckHealth(ctx, req)
return requestStatusFromError(innerErr), innerErr
return instrumentationutils.RequestStatusFromError(innerErr), innerErr
})
return resp, err
@@ -122,9 +125,98 @@ func (m *LoggerMiddleware) CollectMetrics(ctx context.Context, req *backend.Coll
}
var resp *backend.CollectMetricsResult
err := m.logRequest(ctx, func(ctx context.Context) (status requestStatus, innerErr error) {
err := m.logRequest(ctx, func(ctx context.Context) (instrumentationutils.RequestStatus, error) {
var innerErr error
resp, innerErr = m.next.CollectMetrics(ctx, req)
return requestStatusFromError(innerErr), innerErr
return instrumentationutils.RequestStatusFromError(innerErr), innerErr
})
return resp, err
}
func (m *LoggerMiddleware) SubscribeStream(ctx context.Context, req *backend.SubscribeStreamRequest) (*backend.SubscribeStreamResponse, error) {
if req == nil {
return m.next.SubscribeStream(ctx, req)
}
var resp *backend.SubscribeStreamResponse
err := m.logRequest(ctx, func(ctx context.Context) (instrumentationutils.RequestStatus, error) {
var innerErr error
resp, innerErr = m.next.SubscribeStream(ctx, req)
return instrumentationutils.RequestStatusFromError(innerErr), innerErr
})
return resp, err
}
func (m *LoggerMiddleware) PublishStream(ctx context.Context, req *backend.PublishStreamRequest) (*backend.PublishStreamResponse, error) {
if req == nil {
return m.next.PublishStream(ctx, req)
}
var resp *backend.PublishStreamResponse
err := m.logRequest(ctx, func(ctx context.Context) (instrumentationutils.RequestStatus, error) {
var innerErr error
resp, innerErr = m.next.PublishStream(ctx, req)
return instrumentationutils.RequestStatusFromError(innerErr), innerErr
})
return resp, err
}
func (m *LoggerMiddleware) RunStream(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
if req == nil {
return m.next.RunStream(ctx, req, sender)
}
err := m.logRequest(ctx, func(ctx context.Context) (instrumentationutils.RequestStatus, error) {
innerErr := m.next.RunStream(ctx, req, sender)
return instrumentationutils.RequestStatusFromError(innerErr), innerErr
})
return err
}
func (m *LoggerMiddleware) ValidateAdmission(ctx context.Context, req *backend.AdmissionRequest) (*backend.ValidationResponse, error) {
if req == nil {
return m.next.ValidateAdmission(ctx, req)
}
var resp *backend.ValidationResponse
err := m.logRequest(ctx, func(ctx context.Context) (instrumentationutils.RequestStatus, error) {
var innerErr error
resp, innerErr = m.next.ValidateAdmission(ctx, req)
return instrumentationutils.RequestStatusFromError(innerErr), innerErr
})
return resp, err
}
func (m *LoggerMiddleware) MutateAdmission(ctx context.Context, req *backend.AdmissionRequest) (*backend.MutationResponse, error) {
if req == nil {
return m.next.MutateAdmission(ctx, req)
}
var resp *backend.MutationResponse
err := m.logRequest(ctx, func(ctx context.Context) (instrumentationutils.RequestStatus, error) {
var innerErr error
resp, innerErr = m.next.MutateAdmission(ctx, req)
return instrumentationutils.RequestStatusFromError(innerErr), innerErr
})
return resp, err
}
func (m *LoggerMiddleware) ConvertObject(ctx context.Context, req *backend.ConversionRequest) (*backend.ConversionResponse, error) {
if req == nil {
return m.next.ConvertObject(ctx, req)
}
var resp *backend.ConversionResponse
err := m.logRequest(ctx, func(ctx context.Context) (instrumentationutils.RequestStatus, error) {
var innerErr error
resp, innerErr = m.next.ConvertObject(ctx, req)
return instrumentationutils.RequestStatusFromError(innerErr), innerErr
})
return resp, err
pkg/services/pluginsintegration/clientmiddleware/metrics_middleware.go
+74 -37
View File
@@ -9,6 +9,7 @@ import (
"github.com/grafana/grafana/pkg/infra/tracing"
"github.com/grafana/grafana/pkg/plugins"
"github.com/grafana/grafana/pkg/plugins/instrumentationutils"
"github.com/grafana/grafana/pkg/plugins/manager/registry"
"github.com/grafana/grafana/pkg/plugins/pluginrequestmeta"
)
@@ -94,17 +95,18 @@ func (m *MetricsMiddleware) pluginTarget(ctx context.Context, pluginID, pluginVe
}
// instrumentPluginRequestSize tracks the size of the given request in the m.pluginRequestSize metric.
func (m *MetricsMiddleware) instrumentPluginRequestSize(ctx context.Context, pluginCtx backend.PluginContext, endpoint string, requestSize float64) error {
func (m *MetricsMiddleware) instrumentPluginRequestSize(ctx context.Context, pluginCtx backend.PluginContext, requestSize float64) error {
target, err := m.pluginTarget(ctx, pluginCtx.PluginID, pluginCtx.PluginVersion)
if err != nil {
return err
}
m.pluginRequestSize.WithLabelValues("grafana-backend", pluginCtx.PluginID, endpoint, target).Observe(requestSize)
endpoint := backend.EndpointFromContext(ctx)
m.pluginRequestSize.WithLabelValues("grafana-backend", pluginCtx.PluginID, string(endpoint), target).Observe(requestSize)
return nil
}
// instrumentPluginRequest increments the m.pluginRequestCounter metric and tracks the duration of the given request.
func (m *MetricsMiddleware) instrumentPluginRequest(ctx context.Context, pluginCtx backend.PluginContext, endpoint string, fn func(context.Context) (requestStatus, error)) error {
func (m *MetricsMiddleware) instrumentPluginRequest(ctx context.Context, pluginCtx backend.PluginContext, fn func(context.Context) (instrumentationutils.RequestStatus, error)) error {
target, err := m.pluginTarget(ctx, pluginCtx.PluginID, pluginCtx.PluginVersion)
if err != nil {
return err
@@ -116,10 +118,11 @@ func (m *MetricsMiddleware) instrumentPluginRequest(ctx context.Context, pluginC
elapsed := time.Since(start)
statusSource := pluginrequestmeta.StatusSourceFromContext(ctx)
endpoint := backend.EndpointFromContext(ctx)
pluginRequestDurationWithLabels := m.pluginRequestDuration.WithLabelValues(pluginCtx.PluginID, endpoint, target, string(statusSource))
pluginRequestCounterWithLabels := m.pluginRequestCounter.WithLabelValues(pluginCtx.PluginID, endpoint, status.String(), target, string(statusSource))
pluginRequestDurationSecondsWithLabels := m.pluginRequestDurationSeconds.WithLabelValues("grafana-backend", pluginCtx.PluginID, endpoint, status.String(), target, string(statusSource))
pluginRequestDurationWithLabels := m.pluginRequestDuration.WithLabelValues(pluginCtx.PluginID, string(endpoint), target, string(statusSource))
pluginRequestCounterWithLabels := m.pluginRequestCounter.WithLabelValues(pluginCtx.PluginID, string(endpoint), status.String(), target, string(statusSource))
pluginRequestDurationSecondsWithLabels := m.pluginRequestDurationSeconds.WithLabelValues("grafana-backend", pluginCtx.PluginID, string(endpoint), status.String(), target, string(statusSource))
if traceID := tracing.TraceIDFromContext(ctx, true); traceID != "" {
pluginRequestDurationWithLabels.(prometheus.ExemplarObserver).ObserveWithExemplar(
@@ -144,74 +147,108 @@ func (m *MetricsMiddleware) QueryData(ctx context.Context, req *backend.QueryDat
requestSize += float64(len(v.JSON))
}
if err := m.instrumentPluginRequestSize(ctx, req.PluginContext, endpointQueryData, requestSize); err != nil {
if err := m.instrumentPluginRequestSize(ctx, req.PluginContext, requestSize); err != nil {
return nil, err
}
var resp *backend.QueryDataResponse
err := m.instrumentPluginRequest(ctx, req.PluginContext, endpointQueryData, func(ctx context.Context) (status requestStatus, innerErr error) {
err := m.instrumentPluginRequest(ctx, req.PluginContext, func(ctx context.Context) (instrumentationutils.RequestStatus, error) {
var innerErr error
resp, innerErr = m.next.QueryData(ctx, req)
return requestStatusFromQueryDataResponse(resp, innerErr), innerErr
return instrumentationutils.RequestStatusFromQueryDataResponse(resp, innerErr), innerErr
})
return resp, err
}
func (m *MetricsMiddleware) CallResource(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error {
if err := m.instrumentPluginRequestSize(ctx, req.PluginContext, endpointCallResource, float64(len(req.Body))); err != nil {
if err := m.instrumentPluginRequestSize(ctx, req.PluginContext, float64(len(req.Body))); err != nil {
return err
}
return m.instrumentPluginRequest(ctx, req.PluginContext, endpointCallResource, func(ctx context.Context) (requestStatus, error) {
return m.instrumentPluginRequest(ctx, req.PluginContext, func(ctx context.Context) (instrumentationutils.RequestStatus, error) {
innerErr := m.next.CallResource(ctx, req, sender)
return requestStatusFromError(innerErr), innerErr
return instrumentationutils.RequestStatusFromError(innerErr), innerErr
})
}
func (m *MetricsMiddleware) CheckHealth(ctx context.Context, req *backend.CheckHealthRequest) (*backend.CheckHealthResult, error) {
var result *backend.CheckHealthResult
err := m.instrumentPluginRequest(ctx, req.PluginContext, endpointCheckHealth, func(ctx context.Context) (status requestStatus, innerErr error) {
result, innerErr = m.next.CheckHealth(ctx, req)
return requestStatusFromError(innerErr), innerErr
var resp *backend.CheckHealthResult
err := m.instrumentPluginRequest(ctx, req.PluginContext, func(ctx context.Context) (instrumentationutils.RequestStatus, error) {
var innerErr error
resp, innerErr = m.next.CheckHealth(ctx, req)
return instrumentationutils.RequestStatusFromError(innerErr), innerErr
})
return result, err
return resp, err
}
func (m *MetricsMiddleware) CollectMetrics(ctx context.Context, req *backend.CollectMetricsRequest) (*backend.CollectMetricsResult, error) {
var result *backend.CollectMetricsResult
err := m.instrumentPluginRequest(ctx, req.PluginContext, endpointCollectMetrics, func(ctx context.Context) (status requestStatus, innerErr error) {
result, innerErr = m.next.CollectMetrics(ctx, req)
return requestStatusFromError(innerErr), innerErr
var resp *backend.CollectMetricsResult
err := m.instrumentPluginRequest(ctx, req.PluginContext, func(ctx context.Context) (instrumentationutils.RequestStatus, error) {
var innerErr error
resp, innerErr = m.next.CollectMetrics(ctx, req)
return instrumentationutils.RequestStatusFromError(innerErr), innerErr
})
return result, err
return resp, err
}
func (m *MetricsMiddleware) SubscribeStream(ctx context.Context, req *backend.SubscribeStreamRequest) (*backend.SubscribeStreamResponse, error) {
var resp *backend.SubscribeStreamResponse
err := m.instrumentPluginRequest(ctx, req.PluginContext, func(ctx context.Context) (instrumentationutils.RequestStatus, error) {
var innerErr error
resp, innerErr = m.next.SubscribeStream(ctx, req)
return instrumentationutils.RequestStatusFromError(innerErr), innerErr
})
return resp, err
}
func (m *MetricsMiddleware) PublishStream(ctx context.Context, req *backend.PublishStreamRequest) (*backend.PublishStreamResponse, error) {
var resp *backend.PublishStreamResponse
err := m.instrumentPluginRequest(ctx, req.PluginContext, func(ctx context.Context) (instrumentationutils.RequestStatus, error) {
var innerErr error
resp, innerErr = m.next.PublishStream(ctx, req)
return instrumentationutils.RequestStatusFromError(innerErr), innerErr
})
return resp, err
}
func (m *MetricsMiddleware) RunStream(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
err := m.instrumentPluginRequest(ctx, req.PluginContext, func(ctx context.Context) (instrumentationutils.RequestStatus, error) {
innerErr := m.next.RunStream(ctx, req, sender)
return instrumentationutils.RequestStatusFromError(innerErr), innerErr
})
return err
}
func (m *MetricsMiddleware) ValidateAdmission(ctx context.Context, req *backend.AdmissionRequest) (*backend.ValidationResponse, error) {
var result *backend.ValidationResponse
err := m.instrumentPluginRequest(ctx, req.PluginContext, endpointMutateAdmission, func(ctx context.Context) (status requestStatus, innerErr error) {
result, innerErr = m.next.ValidateAdmission(ctx, req)
return requestStatusFromError(innerErr), innerErr
var resp *backend.ValidationResponse
err := m.instrumentPluginRequest(ctx, req.PluginContext, func(ctx context.Context) (instrumentationutils.RequestStatus, error) {
var innerErr error
resp, innerErr = m.next.ValidateAdmission(ctx, req)
return instrumentationutils.RequestStatusFromError(innerErr), innerErr
})
return result, err
return resp, err
}
func (m *MetricsMiddleware) MutateAdmission(ctx context.Context, req *backend.AdmissionRequest) (*backend.MutationResponse, error) {
var result *backend.MutationResponse
err := m.instrumentPluginRequest(ctx, req.PluginContext, endpointMutateAdmission, func(ctx context.Context) (status requestStatus, innerErr error) {
result, innerErr = m.next.MutateAdmission(ctx, req)
return requestStatusFromError(innerErr), innerErr
var resp *backend.MutationResponse
err := m.instrumentPluginRequest(ctx, req.PluginContext, func(ctx context.Context) (instrumentationutils.RequestStatus, error) {
var innerErr error
resp, innerErr = m.next.MutateAdmission(ctx, req)
return instrumentationutils.RequestStatusFromError(innerErr), innerErr
})
return result, err
return resp, err
}
func (m *MetricsMiddleware) ConvertObject(ctx context.Context, req *backend.ConversionRequest) (*backend.ConversionResponse, error) {
var result *backend.ConversionResponse
err := m.instrumentPluginRequest(ctx, req.PluginContext, endpointMutateAdmission, func(ctx context.Context) (status requestStatus, innerErr error) {
result, innerErr = m.next.ConvertObject(ctx, req)
return requestStatusFromError(innerErr), innerErr
var resp *backend.ConversionResponse
err := m.instrumentPluginRequest(ctx, req.PluginContext, func(ctx context.Context) (instrumentationutils.RequestStatus, error) {
var innerErr error
resp, innerErr = m.next.ConvertObject(ctx, req)
return instrumentationutils.RequestStatusFromError(innerErr), innerErr
})
return result, err
return resp, err
}
pkg/services/pluginsintegration/clientmiddleware/metrics_middleware_test.go
+12 -11
View File
@@ -14,6 +14,7 @@ import (
"github.com/grafana/grafana/pkg/plugins"
"github.com/grafana/grafana/pkg/plugins/backendplugin"
"github.com/grafana/grafana/pkg/plugins/instrumentationutils"
"github.com/grafana/grafana/pkg/plugins/manager/client/clienttest"
"github.com/grafana/grafana/pkg/plugins/manager/fakes"
"github.com/grafana/grafana/pkg/plugins/pluginrequestmeta"
@@ -32,12 +33,12 @@ func TestInstrumentationMiddleware(t *testing.T) {
pCtx := backend.PluginContext{PluginID: pluginID}
t.Run("should instrument requests", func(t *testing.T) {
for _, tc := range []struct {
expEndpoint string
expEndpoint backend.Endpoint
fn func(cdt *clienttest.ClientDecoratorTest) error
shouldInstrumentRequestSize bool
}{
{
expEndpoint: endpointCheckHealth,
expEndpoint: backend.EndpointCheckHealth,
fn: func(cdt *clienttest.ClientDecoratorTest) error {
_, err := cdt.Decorator.CheckHealth(context.Background(), &backend.CheckHealthRequest{PluginContext: pCtx})
return err
@@ -45,14 +46,14 @@ func TestInstrumentationMiddleware(t *testing.T) {
shouldInstrumentRequestSize: false,
},
{
expEndpoint: endpointCallResource,
expEndpoint: backend.EndpointCallResource,
fn: func(cdt *clienttest.ClientDecoratorTest) error {
return cdt.Decorator.CallResource(context.Background(), &backend.CallResourceRequest{PluginContext: pCtx}, nopCallResourceSender)
},
shouldInstrumentRequestSize: true,
},
{
expEndpoint: endpointQueryData,
expEndpoint: backend.EndpointQueryData,
fn: func(cdt *clienttest.ClientDecoratorTest) error {
_, err := cdt.Decorator.QueryData(context.Background(), &backend.QueryDataRequest{PluginContext: pCtx})
return err
@@ -60,7 +61,7 @@ func TestInstrumentationMiddleware(t *testing.T) {
shouldInstrumentRequestSize: true,
},
{
expEndpoint: endpointCollectMetrics,
expEndpoint: backend.EndpointCollectMetrics,
fn: func(cdt *clienttest.ClientDecoratorTest) error {
_, err := cdt.Decorator.CollectMetrics(context.Background(), &backend.CollectMetricsRequest{PluginContext: pCtx})
return err
@@ -68,7 +69,7 @@ func TestInstrumentationMiddleware(t *testing.T) {
shouldInstrumentRequestSize: false,
},
} {
t.Run(tc.expEndpoint, func(t *testing.T) {
t.Run(string(tc.expEndpoint), func(t *testing.T) {
promRegistry := prometheus.NewRegistry()
pluginsRegistry := fakes.NewFakePluginRegistry()
require.NoError(t, pluginsRegistry.Add(context.Background(), &plugins.Plugin{
@@ -89,12 +90,12 @@ func TestInstrumentationMiddleware(t *testing.T) {
require.Equal(t, 1, testutil.CollectAndCount(promRegistry, metricRequestDurationMs))
require.Equal(t, 1, testutil.CollectAndCount(promRegistry, metricRequestDurationS))
counter := mw.pluginMetrics.pluginRequestCounter.WithLabelValues(pluginID, tc.expEndpoint, requestStatusOK.String(), string(backendplugin.TargetUnknown), string(pluginrequestmeta.DefaultStatusSource))
counter := mw.pluginMetrics.pluginRequestCounter.WithLabelValues(pluginID, string(tc.expEndpoint), instrumentationutils.RequestStatusOK.String(), string(backendplugin.TargetUnknown), string(pluginrequestmeta.DefaultStatusSource))
require.Equal(t, 1.0, testutil.ToFloat64(counter))
for _, m := range []string{metricRequestDurationMs, metricRequestDurationS} {
require.NoError(t, checkHistogram(promRegistry, m, map[string]string{
"plugin_id": pluginID,
"endpoint": tc.expEndpoint,
"endpoint": string(tc.expEndpoint),
"target": string(backendplugin.TargetUnknown),
}))
}
@@ -102,7 +103,7 @@ func TestInstrumentationMiddleware(t *testing.T) {
require.Equal(t, 1, testutil.CollectAndCount(promRegistry, metricRequestSize), "request size should have been instrumented")
require.NoError(t, checkHistogram(promRegistry, metricRequestSize, map[string]string{
"plugin_id": pluginID,
"endpoint": tc.expEndpoint,
"endpoint": string(tc.expEndpoint),
"target": string(backendplugin.TargetUnknown),
"source": "grafana-backend",
}), "request size should have been instrumented")
@@ -116,8 +117,8 @@ func TestInstrumentationMiddlewareStatusSource(t *testing.T) {
const labelStatusSource = "status_source"
queryDataErrorCounterLabels := prometheus.Labels{
"plugin_id": pluginID,
"endpoint": endpointQueryData,
"status": requestStatusError.String(),
"endpoint": string(backend.EndpointQueryData),
"status": instrumentationutils.RequestStatusError.String(),
"target": string(backendplugin.TargetUnknown),
}
downstreamErrorResponse := backend.DataResponse{
pkg/services/pluginsintegration/clientmiddleware/resource_response_middleware.go
+1 -1
View File
@@ -35,7 +35,7 @@ func (m *ResourceResponseMiddleware) CallResource(ctx context.Context, req *back
}
processedStreams := 0
wrappedSender := callResourceResponseSenderFunc(func(res *backend.CallResourceResponse) error {
wrappedSender := backend.CallResourceResponseSenderFunc(func(res *backend.CallResourceResponse) error {
if processedStreams == 0 {
if res.Headers == nil {
res.Headers = map[string][]string{}
pkg/services/pluginsintegration/clientmiddleware/resource_response_middleware_test.go
+1 -1
View File
@@ -24,7 +24,7 @@ func TestResourceResponseMiddleware(t *testing.T) {
)
var sentResponse *backend.CallResourceResponse
sender := callResourceResponseSenderFunc(func(res *backend.CallResourceResponse) error {
sender := backend.CallResourceResponseSenderFunc(func(res *backend.CallResourceResponse) error {
sentResponse = res
return nil
})
pkg/services/pluginsintegration/clientmiddleware/testing.go
+1 -1
View File
@@ -2,6 +2,6 @@ package clientmiddleware
import "github.com/grafana/grafana-plugin-sdk-go/backend"
var nopCallResourceSender = callResourceResponseSenderFunc(func(res *backend.CallResourceResponse) error {
var nopCallResourceSender = backend.CallResourceResponseSenderFunc(func(res *backend.CallResourceResponse) error {
return nil
})
pkg/services/pluginsintegration/clientmiddleware/tracing_middleware.go
+13 -13
View File
@@ -44,10 +44,10 @@ func setSpanAttributeFromHTTPHeader(headers http.Header, span trace.Span, attrib
// plugin id, org id, user login, ds, dashboard and panel info. The second function returned is a cleanup function,
// which should be called by the caller (deferred) and will set the span status/error and end the span.
func (m *TracingMiddleware) traceWrap(
ctx context.Context, pluginContext backend.PluginContext, opName string,
ctx context.Context, pluginContext backend.PluginContext,
) (context.Context, func(error)) {
// Start span
ctx, span := m.tracer.Start(ctx, "PluginClient."+opName, trace.WithAttributes(
endpoint := backend.EndpointFromContext(ctx)
ctx, span := m.tracer.Start(ctx, "PluginClient."+string(endpoint), trace.WithAttributes(
// Attach some plugin context information to span
attribute.String("plugin_id", pluginContext.PluginID),
attribute.Int64("org_id", pluginContext.OrgID),
@@ -82,7 +82,7 @@ func (m *TracingMiddleware) traceWrap(
func (m *TracingMiddleware) QueryData(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
var err error
ctx, end := m.traceWrap(ctx, req.PluginContext, "queryData")
ctx, end := m.traceWrap(ctx, req.PluginContext)
defer func() { end(err) }()
resp, err := m.next.QueryData(ctx, req)
return resp, err
@@ -90,7 +90,7 @@ func (m *TracingMiddleware) QueryData(ctx context.Context, req *backend.QueryDat
func (m *TracingMiddleware) CallResource(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error {
var err error
ctx, end := m.traceWrap(ctx, req.PluginContext, "callResource")
ctx, end := m.traceWrap(ctx, req.PluginContext)
defer func() { end(err) }()
err = m.next.CallResource(ctx, req, sender)
return err
@@ -98,7 +98,7 @@ func (m *TracingMiddleware) CallResource(ctx context.Context, req *backend.CallR
func (m *TracingMiddleware) CheckHealth(ctx context.Context, req *backend.CheckHealthRequest) (*backend.CheckHealthResult, error) {
var err error
ctx, end := m.traceWrap(ctx, req.PluginContext, "checkHealth")
ctx, end := m.traceWrap(ctx, req.PluginContext)
defer func() { end(err) }()
resp, err := m.next.CheckHealth(ctx, req)
return resp, err
@@ -106,7 +106,7 @@ func (m *TracingMiddleware) CheckHealth(ctx context.Context, req *backend.CheckH
func (m *TracingMiddleware) CollectMetrics(ctx context.Context, req *backend.CollectMetricsRequest) (*backend.CollectMetricsResult, error) {
var err error
ctx, end := m.traceWrap(ctx, req.PluginContext, "collectMetrics")
ctx, end := m.traceWrap(ctx, req.PluginContext)
defer func() { end(err) }()
resp, err := m.next.CollectMetrics(ctx, req)
return resp, err
@@ -114,7 +114,7 @@ func (m *TracingMiddleware) CollectMetrics(ctx context.Context, req *backend.Col
func (m *TracingMiddleware) SubscribeStream(ctx context.Context, req *backend.SubscribeStreamRequest) (*backend.SubscribeStreamResponse, error) {
var err error
ctx, end := m.traceWrap(ctx, req.PluginContext, "subscribeStream")
ctx, end := m.traceWrap(ctx, req.PluginContext)
defer func() { end(err) }()
resp, err := m.next.SubscribeStream(ctx, req)
return resp, err
@@ -122,7 +122,7 @@ func (m *TracingMiddleware) SubscribeStream(ctx context.Context, req *backend.Su
func (m *TracingMiddleware) PublishStream(ctx context.Context, req *backend.PublishStreamRequest) (*backend.PublishStreamResponse, error) {
var err error
ctx, end := m.traceWrap(ctx, req.PluginContext, "publishStream")
ctx, end := m.traceWrap(ctx, req.PluginContext)
defer func() { end(err) }()
resp, err := m.next.PublishStream(ctx, req)
return resp, err
@@ -130,7 +130,7 @@ func (m *TracingMiddleware) PublishStream(ctx context.Context, req *backend.Publ
func (m *TracingMiddleware) RunStream(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
var err error
ctx, end := m.traceWrap(ctx, req.PluginContext, "runStream")
ctx, end := m.traceWrap(ctx, req.PluginContext)
defer func() { end(err) }()
err = m.next.RunStream(ctx, req, sender)
return err
@@ -139,7 +139,7 @@ func (m *TracingMiddleware) RunStream(ctx context.Context, req *backend.RunStrea
// ValidateAdmission implements backend.AdmissionHandler.
func (m *TracingMiddleware) ValidateAdmission(ctx context.Context, req *backend.AdmissionRequest) (*backend.ValidationResponse, error) {
var err error
ctx, end := m.traceWrap(ctx, req.PluginContext, endpointValidateAdmission)
ctx, end := m.traceWrap(ctx, req.PluginContext)
defer func() { end(err) }()
resp, err := m.next.ValidateAdmission(ctx, req)
return resp, err
@@ -148,7 +148,7 @@ func (m *TracingMiddleware) ValidateAdmission(ctx context.Context, req *backend.
// MutateAdmission implements backend.AdmissionHandler.
func (m *TracingMiddleware) MutateAdmission(ctx context.Context, req *backend.AdmissionRequest) (*backend.MutationResponse, error) {
var err error
ctx, end := m.traceWrap(ctx, req.PluginContext, endpointMutateAdmission)
ctx, end := m.traceWrap(ctx, req.PluginContext)
defer func() { end(err) }()
resp, err := m.next.MutateAdmission(ctx, req)
return resp, err
@@ -157,7 +157,7 @@ func (m *TracingMiddleware) MutateAdmission(ctx context.Context, req *backend.Ad
// ConvertObject implements backend.AdmissionHandler.
func (m *TracingMiddleware) ConvertObject(ctx context.Context, req *backend.ConversionRequest) (*backend.ConversionResponse, error) {
var err error
ctx, end := m.traceWrap(ctx, req.PluginContext, endpointConvertObject)
ctx, end := m.traceWrap(ctx, req.PluginContext)
defer func() { end(err) }()
resp, err := m.next.ConvertObject(ctx, req)
return resp, err
pkg/services/pluginsintegration/clientmiddleware/utils.go
-81
View File
@@ -1,81 +0,0 @@
package clientmiddleware
import (
"context"
"errors"
"github.com/grafana/grafana-plugin-sdk-go/backend"
)
type requestStatus int
const (
requestStatusOK requestStatus = iota
requestStatusCancelled
requestStatusError
)
func (status requestStatus) String() string {
names := [...]string{"ok", "cancelled", "error"}
if status < requestStatusOK || status > requestStatusError {
return ""
}
return names[status]
}
const (
endpointCallResource = "callResource"
endpointCheckHealth = "checkHealth"
endpointCollectMetrics = "collectMetrics"
endpointQueryData = "queryData"
endpointSubscribeStream = "subscribeStream"
endpointPublishStream = "publishStream"
endpointRunStream = "runStream"
endpointValidateAdmission = "validateAdmission"
endpointMutateAdmission = "mutateAdmission"
endpointConvertObject = "convertObject"
)
type callResourceResponseSenderFunc func(res *backend.CallResourceResponse) error
func (fn callResourceResponseSenderFunc) Send(res *backend.CallResourceResponse) error {
return fn(res)
}
func requestStatusFromError(err error) requestStatus {
status := requestStatusOK
if err != nil {
status = requestStatusError
if errors.Is(err, context.Canceled) {
status = requestStatusCancelled
}
}
return status
}
func requestStatusFromQueryDataResponse(res *backend.QueryDataResponse, err error) requestStatus {
if err != nil {
return requestStatusFromError(err)
}
status := requestStatusOK
if res != nil {
for _, dr := range res.Responses {
if dr.Error != nil {
s := requestStatusFromError(dr.Error)
if s > status {
status = s
}
if status == requestStatusError {
break
}
}
}
}
return status
}
pkg/services/pluginsintegration/clientmiddleware/utils_test.go
-137
View File
@@ -1,137 +0,0 @@
package clientmiddleware
import (
"context"
"errors"
"fmt"
"testing"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana-plugin-sdk-go/data"
"github.com/stretchr/testify/require"
)
func TestRequestStatus(t *testing.T) {
tcs := []struct {
s requestStatus
expectedLabel string
}{
{
s: requestStatusOK,
expectedLabel: "ok",
},
{
s: requestStatusError,
expectedLabel: "error",
},
{
s: requestStatusCancelled,
expectedLabel: "cancelled",
},
}
for _, tc := range tcs {
t.Run(tc.s.String(), func(t *testing.T) {
require.Equal(t, tc.expectedLabel, tc.s.String())
require.Equal(t, tc.expectedLabel, fmt.Sprint(tc.s))
})
}
}
func TestRequestStatusFromError(t *testing.T) {
tcs := []struct {
desc string
err error
expectedStatus requestStatus
}{
{
desc: "no error should be status ok",
err: nil,
expectedStatus: requestStatusOK,
},
{
desc: "error should be status error",
err: errors.New("boom"),
expectedStatus: requestStatusError,
},
{
desc: "context canceled should be status cancelled",
err: context.Canceled,
expectedStatus: requestStatusCancelled,
},
}
for _, tc := range tcs {
t.Run(tc.desc, func(t *testing.T) {
status := requestStatusFromError(tc.err)
require.Equal(t, tc.expectedStatus, status)
})
}
}
func TestRequestStatusFromQueryDataResponse(t *testing.T) {
responseWithoutError := backend.NewQueryDataResponse()
responseWithoutError.Responses["A"] = backend.DataResponse{
Frames: data.Frames{data.NewFrame("test")},
}
responseWithError := backend.NewQueryDataResponse()
responseWithError.Responses["A"] = backend.DataResponse{
Error: errors.New("boom"),
}
responseWithMultipleErrors := backend.NewQueryDataResponse()
responseWithMultipleErrors.Responses["A"] = backend.DataResponse{
Error: context.Canceled,
}
responseWithMultipleErrors.Responses["B"] = backend.DataResponse{
Frames: data.Frames{data.NewFrame("test")},
}
responseWithMultipleErrors.Responses["C"] = backend.DataResponse{
Error: errors.New("boom"),
}
tcs := []struct {
desc string
resp *backend.QueryDataResponse
err error
expectedStatus requestStatus
}{
{
desc: "no error should be status ok",
err: nil,
expectedStatus: requestStatusOK,
},
{
desc: "error should be status error",
err: errors.New("boom"),
expectedStatus: requestStatusError,
},
{
desc: "context canceled should be status cancelled",
err: context.Canceled,
expectedStatus: requestStatusCancelled,
},
{
desc: "response without error should be status ok",
resp: responseWithoutError,
expectedStatus: requestStatusOK,
},
{
desc: "response with error should be status error",
resp: responseWithError,
expectedStatus: requestStatusError,
},
{
desc: "response with multiple error should pick the highest status cancelled",
resp: responseWithMultipleErrors,
expectedStatus: requestStatusError,
},
}
for _, tc := range tcs {
t.Run(tc.desc, func(t *testing.T) {
status := requestStatusFromQueryDataResponse(tc.resp, tc.err)
require.Equal(t, tc.expectedStatus, status)
})
}
}