Instrument backend plugin requests (#23346)

pkg/plugins/backendplugin/backend_plugin.go

@@ -141,7 +141,14 @@ func (p *BackendPlugin) CollectMetrics(ctx context.Context) (*pluginv2.CollectMe
 		}, nil
 		}, nil
 	}
 	}
 
 
-	res, err := p.diagnostics.CollectMetrics(ctx, &pluginv2.CollectMetricsRequest{})
+	var res *pluginv2.CollectMetricsResponse
+	err := InstrumentPluginRequest(p.id, "metrics", func() error {
+		var innerErr error
+		res, innerErr = p.diagnostics.CollectMetrics(ctx, &pluginv2.CollectMetricsRequest{})
+
+		return innerErr
+	})
+
 	if err != nil {
 	if err != nil {
 		if st, ok := status.FromError(err); ok {
 		if st, ok := status.FromError(err); ok {
 			if st.Code() == codes.Unimplemented {
 			if st.Code() == codes.Unimplemented {
@@ -197,7 +204,13 @@ func (p *BackendPlugin) checkHealth(ctx context.Context, config *PluginConfig) (
 		}
 		}
 	}
 	}
 
 
-	res, err := p.diagnostics.CheckHealth(ctx, &pluginv2.CheckHealthRequest{Config: pconfig})
+	var res *pluginv2.CheckHealthResponse
+	err = InstrumentPluginRequest(p.id, "checkhealth", func() error {
+		var innerErr error
+		res, innerErr = p.diagnostics.CheckHealth(ctx, &pluginv2.CheckHealthRequest{Config: pconfig})
+		return innerErr
+	})
+
 	if err != nil {
 	if err != nil {
 		if st, ok := status.FromError(err); ok {
 		if st, ok := status.FromError(err); ok {
 			if st.Code() == codes.Unimplemented {
 			if st.Code() == codes.Unimplemented {

pkg/plugins/backendplugin/instrumentation.go

+package backendplugin
+
+import (
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+var (
+	pluginRequestCounter  *prometheus.CounterVec
+	pluginRequestDuration *prometheus.SummaryVec
+)
+
+func init() {
+	pluginRequestCounter = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "grafana",
+		Name:      "plugin_request_total",
+		Help:      "The total amount of plugin requests",
+	}, []string{"plugin_id", "endpoint", "status"})
+
+	pluginRequestDuration = prometheus.NewSummaryVec(prometheus.SummaryOpts{
+		Namespace:  "grafana",
+		Name:       "plugin_request_duration_milliseconds",
+		Help:       "Plugin request duration",
+		Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
+	}, []string{"plugin_id", "endpoint"})
+
+	prometheus.MustRegister(pluginRequestCounter, pluginRequestDuration)
+}
+
+// InstrumentPluginRequest instruments success rate and latency of `fn`
+func InstrumentPluginRequest(pluginID string, endpoint string, fn func() error) error {
+	status := "ok"
+
+	start := time.Now()
+
+	err := fn()
+	if err != nil {
+		status = "error"
+	}
+
+	elapsed := time.Since(start) / time.Millisecond
+	pluginRequestDuration.WithLabelValues(pluginID, endpoint).Observe(float64(elapsed))
+	pluginRequestCounter.WithLabelValues(pluginID, endpoint, status).Inc()
+
+	return err
+}

pkg/plugins/backendplugin/manager.go

@@ -8,6 +8,7 @@ import (
 	"time"
 	"time"
 
 
 	"github.com/grafana/grafana/pkg/models"
 	"github.com/grafana/grafana/pkg/models"
+	"github.com/grafana/grafana/pkg/util/errutil"
 	"github.com/grafana/grafana/pkg/util/proxyutil"
 	"github.com/grafana/grafana/pkg/util/proxyutil"
 
 
 	"github.com/grafana/grafana/pkg/infra/log"
 	"github.com/grafana/grafana/pkg/infra/log"
@@ -187,17 +188,17 @@ func (m *manager) CheckHealth(ctx context.Context, pluginConfig *PluginConfig) (
 }
 }
 
 
 // CallResource calls a plugin resource.
 // CallResource calls a plugin resource.
-func (m *manager) CallResource(config PluginConfig, c *models.ReqContext, path string) {
+func (m *manager) CallResource(config PluginConfig, reqCtx *models.ReqContext, path string) {
 	m.pluginsMu.RLock()
 	m.pluginsMu.RLock()
 	p, registered := m.plugins[config.PluginID]
 	p, registered := m.plugins[config.PluginID]
 	m.pluginsMu.RUnlock()
 	m.pluginsMu.RUnlock()
 
 
 	if !registered {
 	if !registered {
-		c.JsonApiErr(404, "Plugin not registered", nil)
+		reqCtx.JsonApiErr(404, "Plugin not registered", nil)
 		return
 		return
 	}
 	}
 
 
-	clonedReq := c.Req.Clone(c.Req.Context())
+	clonedReq := reqCtx.Req.Clone(reqCtx.Req.Context())
 	keepCookieNames := []string{}
 	keepCookieNames := []string{}
 	if config.JSONData != nil {
 	if config.JSONData != nil {
 		if keepCookies := config.JSONData.Get("keepCookies"); keepCookies != nil {
 		if keepCookies := config.JSONData.Get("keepCookies"); keepCookies != nil {
@@ -208,9 +209,9 @@ func (m *manager) CallResource(config PluginConfig, c *models.ReqContext, path s
 	proxyutil.ClearCookieHeader(clonedReq, keepCookieNames)
 	proxyutil.ClearCookieHeader(clonedReq, keepCookieNames)
 	proxyutil.PrepareProxyRequest(clonedReq)
 	proxyutil.PrepareProxyRequest(clonedReq)
 
 
-	body, err := c.Req.Body().Bytes()
+	body, err := reqCtx.Req.Body().Bytes()
 	if err != nil {
 	if err != nil {
-		c.JsonApiErr(500, "Failed to read request body", err)
+		reqCtx.JsonApiErr(500, "Failed to read request body", err)
 		return
 		return
 	}
 	}
 
 
@@ -221,32 +222,41 @@ func (m *manager) CallResource(config PluginConfig, c *models.ReqContext, path s
 		URL:     clonedReq.URL.String(),
 		URL:     clonedReq.URL.String(),
 		Headers: clonedReq.Header,
 		Headers: clonedReq.Header,
 		Body:    body,
 		Body:    body,
-		User:    c.SignedInUser,
+		User:    reqCtx.SignedInUser,
 	}
 	}
 
 
+	err = InstrumentPluginRequest(p.id, "resource", func() error {
 		stream, err := p.callResource(clonedReq.Context(), req)
 		stream, err := p.callResource(clonedReq.Context(), req)
 		if err != nil {
 		if err != nil {
-		c.JsonApiErr(500, "Failed to call resource", err)
-		return
+			return errutil.Wrap("Failed to call resource", err)
+		}
+
+		return flushStream(p, stream, reqCtx)
+	})
+
+	if err != nil {
+		reqCtx.JsonApiErr(500, "Failed to ", err)
 	}
 	}
+}
 
 
+func flushStream(plugin *BackendPlugin, stream callResourceResultStream, reqCtx *models.ReqContext) error {
 	processedStreams := 0
 	processedStreams := 0
 
 
 	for {
 	for {
 		resp, err := stream.Recv()
 		resp, err := stream.Recv()
 		if err == io.EOF {
 		if err == io.EOF {
 			if processedStreams == 0 {
 			if processedStreams == 0 {
-				c.JsonApiErr(500, "Received empty resource response ", nil)
+				return errors.New("Received empty resource response")
 			}
 			}
-			return
+			return nil
 		}
 		}
 		if err != nil {
 		if err != nil {
 			if processedStreams == 0 {
 			if processedStreams == 0 {
-				c.JsonApiErr(500, "Failed to receive response from resource call", err)
-			} else {
-				p.logger.Error("Failed to receive response from resource call", "error", err)
+				return errutil.Wrap("Failed to receive response from resource call", err)
 			}
 			}
-			return
+
+			plugin.logger.Error("Failed to receive response from resource call", "error", err)
+			return nil
 		}
 		}
 
 
 		// Expected that headers and status are only part of first stream
 		// Expected that headers and status are only part of first stream
@@ -264,18 +274,18 @@ func (m *manager) CallResource(config PluginConfig, c *models.ReqContext, path s
 				}
 				}
 
 
 				for _, v := range values {
 				for _, v := range values {
-					c.Resp.Header().Add(k, v)
+					reqCtx.Resp.Header().Add(k, v)
 				}
 				}
 			}
 			}
 
 
-			c.WriteHeader(resp.Status)
+			reqCtx.WriteHeader(resp.Status)
 		}
 		}
 
 
-		if _, err := c.Write(resp.Body); err != nil {
-			p.logger.Error("Failed to write resource response", "error", err)
+		if _, err := reqCtx.Write(resp.Body); err != nil {
+			plugin.logger.Error("Failed to write resource response", "error", err)
 		}
 		}
 
 
-		c.Resp.Flush()
+		reqCtx.Resp.Flush()
 		processedStreams++
 		processedStreams++
 	}
 	}
 }
 }

pkg/plugins/datasource/wrapper/datasource_plugin_wrapper_v2.go

@@ -77,7 +77,14 @@ func (tw *DatasourcePluginWrapperV2) Query(ctx context.Context, ds *models.DataS
 		})
 		})
 	}
 	}
 
 
-	pbRes, err := tw.DataPlugin.QueryData(ctx, pbQuery)
+	var pbRes *pluginv2.QueryDataResponse
+	err = backendplugin.InstrumentPluginRequest(ds.Type, "dataquery", func() error {
+		var err error
+		pbRes, err = tw.DataPlugin.QueryData(ctx, pbQuery)
+
+		return err
+	})
+
 	if err != nil {
 	if err != nil {
 		return nil, err
 		return nil, err
 	}
 	}