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

Backend plugins: Implement support for resources (#21805)

Browse Source 
Implements initial support for resources using v0.14.0 of SDK.

Ref #21512
This commit is contained in:
Marcus Efraimsson
2020-01-31 11:15:50 +01:00
committed by GitHub
parent 5345868148
commit 0390b5601e
44 changed files with 957 additions and 2195 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/plugins/backendplugin/backend_plugin.go
+49 -6
View File
@@ -5,6 +5,7 @@ import (
"context"
"errors"
"fmt"
"net/http"
"github.com/grafana/grafana-plugin-sdk-go/genproto/pluginv2"
"github.com/prometheus/client_golang/prometheus"
@@ -14,7 +15,6 @@ import (
datasourceV1 "github.com/grafana/grafana-plugin-model/go/datasource"
rendererV1 "github.com/grafana/grafana-plugin-model/go/renderer"
backend "github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana/pkg/infra/log"
"github.com/grafana/grafana/pkg/plugins/backendplugin/collector"
"github.com/grafana/grafana/pkg/util/errutil"
@@ -31,7 +31,8 @@ type BackendPlugin struct {
client *plugin.Client
logger log.Logger
startFns PluginStartFuncs
diagnostics backend.DiagnosticsPlugin
diagnostics DiagnosticsPlugin
core CorePlugin
}
func (p *BackendPlugin) start(ctx context.Context) error {
@@ -61,20 +62,21 @@ func (p *BackendPlugin) start(ctx context.Context) error {
}
if rawDiagnostics != nil {
if plugin, ok := rawDiagnostics.(backend.DiagnosticsPlugin); ok {
if plugin, ok := rawDiagnostics.(DiagnosticsPlugin); ok {
p.diagnostics = plugin
}
}
client = &Client{}
if rawBackend != nil {
if plugin, ok := rawBackend.(backend.BackendPlugin); ok {
client.BackendPlugin = plugin
if plugin, ok := rawBackend.(CorePlugin); ok {
p.core = plugin
client.DatasourcePlugin = plugin
}
}
if rawTransform != nil {
if plugin, ok := rawTransform.(backend.TransformPlugin); ok {
if plugin, ok := rawTransform.(TransformPlugin); ok {
client.TransformPlugin = plugin
}
}
@@ -194,6 +196,47 @@ func (p *BackendPlugin) checkHealth(ctx context.Context) (*pluginv2.CheckHealth_
return res, nil
}
func (p *BackendPlugin) callResource(ctx context.Context, req CallResourceRequest) (*CallResourceResult, error) {
p.logger.Debug("Calling resource", "path", req.Path, "method", req.Method)
reqHeaders := map[string]*pluginv2.CallResource_StringList{}
for k, v := range req.Headers {
reqHeaders[k] = &pluginv2.CallResource_StringList{Values: v}
}
protoReq := &pluginv2.CallResource_Request{
Config: &pluginv2.PluginConfig{},
Path: req.Path,
Method: req.Method,
Url: req.URL,
Headers: reqHeaders,
Body: req.Body,
}
protoResp, err := p.core.CallResource(ctx, protoReq)
if err != nil {
if st, ok := status.FromError(err); ok {
if st.Code() == codes.Unimplemented {
return &CallResourceResult{
Status: http.StatusNotImplemented,
}, nil
}
}
return nil, errutil.Wrap("Failed to call resource", err)
}
respHeaders := map[string][]string{}
for key, values := range protoResp.Headers {
respHeaders[key] = values.Values
}
return &CallResourceResult{
Headers: respHeaders,
Body: protoResp.Body,
Status: int(protoResp.Code),
}, nil
}
// convertMetricFamily converts metric family to prometheus.Metric.
// Copied from https://github.com/prometheus/node_exporter/blob/3ddc82c2d8d11eec53ed5faa8db969a1bb81f8bb/collector/textfile.go#L66-L165
func convertMetricFamily(pluginID string, metricFamily *dto.MetricFamily, ch chan<- prometheus.Metric, logger log.Logger) {
pkg/plugins/backendplugin/client.go
+42 -18
View File
@@ -1,14 +1,17 @@
package backendplugin
import (
"context"
"os/exec"
"github.com/grafana/grafana-plugin-sdk-go/backend/plugin"
"github.com/grafana/grafana/pkg/infra/log"
datasourceV1 "github.com/grafana/grafana-plugin-model/go/datasource"
rendererV1 "github.com/grafana/grafana-plugin-model/go/renderer"
backend "github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/hashicorp/go-plugin"
"github.com/grafana/grafana-plugin-sdk-go/genproto/pluginv2"
goplugin "github.com/hashicorp/go-plugin"
)
const (
@@ -19,24 +22,24 @@ const (
)
// Handshake is the HandshakeConfig used to configure clients and servers.
var handshake = plugin.HandshakeConfig{
var handshake = goplugin.HandshakeConfig{
// The ProtocolVersion is the version that must match between Grafana core
// and Grafana plugins. This should be bumped whenever a (breaking) change
// happens in one or the other that makes it so that they can't safely communicate.
ProtocolVersion: DefaultProtocolVersion,
// The magic cookie values should NEVER be changed.
MagicCookieKey: backend.MagicCookieKey,
MagicCookieValue: backend.MagicCookieValue,
MagicCookieKey: plugin.MagicCookieKey,
MagicCookieValue: plugin.MagicCookieValue,
}
func newClientConfig(executablePath string, logger log.Logger, versionedPlugins map[int]plugin.PluginSet) *plugin.ClientConfig {
return &plugin.ClientConfig{
func newClientConfig(executablePath string, logger log.Logger, versionedPlugins map[int]goplugin.PluginSet) *goplugin.ClientConfig {
return &goplugin.ClientConfig{
Cmd: exec.Command(executablePath),
HandshakeConfig: handshake,
VersionedPlugins: versionedPlugins,
Logger: logWrapper{Logger: logger},
AllowedProtocols: []plugin.Protocol{plugin.ProtocolGRPC},
AllowedProtocols: []goplugin.Protocol{goplugin.ProtocolGRPC},
}
}
@@ -57,7 +60,7 @@ type PluginDescriptor struct {
pluginID string
executablePath string
managed bool
versionedPlugins map[int]plugin.PluginSet
versionedPlugins map[int]goplugin.PluginSet
startFns PluginStartFuncs
}
@@ -68,14 +71,14 @@ func NewBackendPluginDescriptor(pluginID, executablePath string, startFns Plugin
pluginID: pluginID,
executablePath: executablePath,
managed: true,
versionedPlugins: map[int]plugin.PluginSet{
versionedPlugins: map[int]goplugin.PluginSet{
DefaultProtocolVersion: {
pluginID: &datasourceV1.DatasourcePluginImpl{},
},
backend.ProtocolVersion: {
"diagnostics": &backend.DiagnosticsGRPCPlugin{},
"backend": &backend.CoreGRPCPlugin{},
"transform": &backend.TransformGRPCPlugin{},
plugin.ProtocolVersion: {
"diagnostics": &plugin.DiagnosticsGRPCPlugin{},
"backend": &plugin.CoreGRPCPlugin{},
"transform": &plugin.TransformGRPCPlugin{},
},
},
startFns: startFns,
@@ -89,7 +92,7 @@ func NewRendererPluginDescriptor(pluginID, executablePath string, startFns Plugi
pluginID: pluginID,
executablePath: executablePath,
managed: false,
versionedPlugins: map[int]plugin.PluginSet{
versionedPlugins: map[int]goplugin.PluginSet{
DefaultProtocolVersion: {
pluginID: &rendererV1.RendererPluginImpl{},
},
@@ -98,6 +101,28 @@ func NewRendererPluginDescriptor(pluginID, executablePath string, startFns Plugi
}
}
type DiagnosticsPlugin interface {
CollectMetrics(ctx context.Context, req *pluginv2.CollectMetrics_Request) (*pluginv2.CollectMetrics_Response, error)
CheckHealth(ctx context.Context, req *pluginv2.CheckHealth_Request) (*pluginv2.CheckHealth_Response, error)
}
type DatasourcePlugin interface {
DataQuery(ctx context.Context, req *pluginv2.DataQueryRequest) (*pluginv2.DataQueryResponse, error)
}
type CorePlugin interface {
CallResource(ctx context.Context, req *pluginv2.CallResource_Request) (*pluginv2.CallResource_Response, error)
DatasourcePlugin
}
type TransformCallBack interface {
DataQuery(ctx context.Context, req *pluginv2.DataQueryRequest) (*pluginv2.DataQueryResponse, error)
}
type TransformPlugin interface {
DataQuery(ctx context.Context, req *pluginv2.DataQueryRequest, callback TransformCallBack) (*pluginv2.DataQueryResponse, error)
}
// LegacyClient client for communicating with a plugin using the old plugin protocol.
type LegacyClient struct {
DatasourcePlugin datasourceV1.DatasourcePlugin
@@ -106,7 +131,6 @@ type LegacyClient struct {
// Client client for communicating with a plugin using the current plugin protocol.
type Client struct {
DiagnosticsPlugin backend.DiagnosticsPlugin
BackendPlugin backend.BackendPlugin
TransformPlugin backend.TransformPlugin
DatasourcePlugin DatasourcePlugin
TransformPlugin TransformPlugin
}
pkg/plugins/backendplugin/contracts.go
+85
View File
@@ -0,0 +1,85 @@
package backendplugin
import (
"encoding/json"
"strconv"
"github.com/grafana/grafana-plugin-sdk-go/genproto/pluginv2"
)
// HealthStatus is the status of the plugin.
type HealthStatus int
const (
// HealthStatusUnknown means the status of the plugin is unknown.
HealthStatusUnknown HealthStatus = iota
// HealthStatusOk means the status of the plugin is good.
HealthStatusOk
// HealthStatusError means the plugin is in an error state.
HealthStatusError
)
var healthStatusNames = map[int]string{
0: "UNKNOWN",
1: "OK",
2: "ERROR",
}
func (hs HealthStatus) String() string {
s, exists := healthStatusNames[int(hs)]
if exists {
return s
}
return strconv.Itoa(int(hs))
}
// CheckHealthResult check health result.
type CheckHealthResult struct {
Status HealthStatus
Info string
}
func checkHealthResultFromProto(protoResp *pluginv2.CheckHealth_Response) *CheckHealthResult {
status := HealthStatusUnknown
switch protoResp.Status {
case pluginv2.CheckHealth_Response_ERROR:
status = HealthStatusError
case pluginv2.CheckHealth_Response_OK:
status = HealthStatusOk
}
return &CheckHealthResult{
Status: status,
Info: protoResp.Info,
}
}
type PluginInstance struct {
ID int64
Name string
Type string
URL string
JSONData json.RawMessage
}
type PluginConfig struct {
PluginID string
OrgID int64
Instance *PluginInstance
}
type CallResourceRequest struct {
Config PluginConfig
Path string
Method string
URL string
Headers map[string][]string
Body []byte
}
// CallResourceResult call resource result.
type CallResourceResult struct {
Status int
Headers map[string][]string
Body []byte
}
pkg/plugins/backendplugin/manager.go
+43 -25
View File
@@ -6,19 +6,24 @@ import (
"sync"
"time"
"github.com/grafana/grafana-plugin-sdk-go/genproto/pluginv2"
"github.com/prometheus/client_golang/prometheus"
"github.com/grafana/grafana/pkg/api/routing"
"github.com/grafana/grafana/pkg/infra/log"
"github.com/grafana/grafana/pkg/models"
"github.com/grafana/grafana/pkg/plugins/backendplugin/collector"
"github.com/grafana/grafana/pkg/registry"
plugin "github.com/hashicorp/go-plugin"
"golang.org/x/xerrors"
)
var (
// ErrPluginNotRegistered error returned when plugin not registered.
ErrPluginNotRegistered = errors.New("Plugin not registered")
// ErrDiagnosticsNotSupported error returned when plugin doesn't support diagnostics.
ErrDiagnosticsNotSupported = errors.New("Plugin diagnostics not supported")
// ErrHealthCheckFailed error returned when health check failed.
ErrHealthCheckFailed = errors.New("Health check failed")
)
func init() {
registry.Register(&registry.Descriptor{
Name: "BackendPluginManager",
@@ -33,10 +38,13 @@ type Manager interface {
Register(descriptor PluginDescriptor) error
// StartPlugin starts a non-managed backend plugin
StartPlugin(ctx context.Context, pluginID string) error
// CheckHealth checks the health of a registered backend plugin.
CheckHealth(ctx context.Context, pluginID string) (*CheckHealthResult, error)
// CallResource calls a plugin resource.
CallResource(ctx context.Context, req CallResourceRequest) (*CallResourceResult, error)
}
type manager struct {
RouteRegister routing.RouteRegister `inject:""`
pluginsMu sync.RWMutex
plugins map[string]*BackendPlugin
pluginCollector collector.PluginCollector
@@ -49,8 +57,6 @@ func (m *manager) Init() error {
m.pluginCollector = collector.NewPluginCollector()
prometheus.MustRegister(m.pluginCollector)
m.RouteRegister.Get("/api/plugins/:pluginId/health", m.checkHealth)
return nil
}
@@ -140,38 +146,50 @@ func (m *manager) stop() {
}
}
// checkHealth http handler for checking plugin health.
func (m *manager) checkHealth(c *models.ReqContext) {
pluginID := c.Params("pluginId")
// CheckHealth checks the health of a registered backend plugin.
func (m *manager) CheckHealth(ctx context.Context, pluginID string) (*CheckHealthResult, error) {
m.pluginsMu.RLock()
p, registered := m.plugins[pluginID]
m.pluginsMu.RUnlock()
if !registered || !p.supportsDiagnostics() {
c.JsonApiErr(404, "Plugin not found", nil)
return
if !registered {
return nil, ErrPluginNotRegistered
}
res, err := p.checkHealth(c.Req.Context())
if !p.supportsDiagnostics() {
return nil, ErrDiagnosticsNotSupported
}
res, err := p.checkHealth(ctx)
if err != nil {
p.logger.Error("Failed to check plugin health", "error", err)
c.JSON(503, map[string]interface{}{
"status": pluginv2.CheckHealth_Response_ERROR.String(),
})
return
return nil, ErrHealthCheckFailed
}
payload := map[string]interface{}{
"status": res.Status.String(),
"info": res.Info,
return checkHealthResultFromProto(res), nil
}
// CallResource calls a plugin resource.
func (m *manager) CallResource(ctx context.Context, req CallResourceRequest) (*CallResourceResult, error) {
m.pluginsMu.RLock()
p, registered := m.plugins[req.Config.PluginID]
m.pluginsMu.RUnlock()
if !registered {
return nil, ErrPluginNotRegistered
}
if res.Status != pluginv2.CheckHealth_Response_OK {
c.JSON(503, payload)
return
res, err := p.callResource(ctx, req)
if err != nil {
return nil, err
}
c.JSON(200, payload)
// Make sure a content type always is returned in response
if _, exists := res.Headers["Content-Type"]; !exists {
res.Headers["Content-Type"] = []string{"application/json"}
}
return res, nil
}
func startPluginAndRestartKilledProcesses(ctx context.Context, p *BackendPlugin) error {
pkg/plugins/backendplugin/resource/response_writer.go
+137
View File
@@ -0,0 +1,137 @@
package resource
import (
"bytes"
"net/http"
"github.com/grafana/grafana-plugin-sdk-go/genproto/pluginv2"
)
// ResourceResponseWriter is an implementation of http.ResponseWriter that
// records its mutations for later inspection in tests.
type ResourceResponseWriter struct {
// Code is the HTTP response code set by WriteHeader.
//
// Note that if a Handler never calls WriteHeader or Write,
// this might end up being 0, rather than the implicit
// http.StatusOK. To get the implicit value, use the Result
// method.
Code int
// HeaderMap contains the headers explicitly set by the Handler.
// It is an internal detail.
//
// Deprecated: HeaderMap exists for historical compatibility
// and should not be used. To access the headers returned by a handler,
// use the Response.Header map as returned by the Result method.
HeaderMap http.Header
// Body is the buffer to which the Handler's Write calls are sent.
// If nil, the Writes are silently discarded.
Body *bytes.Buffer
// Flushed is whether the Handler called Flush.
Flushed bool
wroteHeader bool
}
// NewResourceResponseWriter returns an initialized ResponseWriter.
func NewResourceResponseWriter() *ResourceResponseWriter {
return &ResourceResponseWriter{
HeaderMap: make(http.Header),
Body: new(bytes.Buffer),
Code: 200,
}
}
// Header implements http.ResponseWriter. It returns the response
// headers to mutate within a handler. To test the headers that were
// written after a handler completes, use the Result method and see
// the returned Response value's Header.
func (rw *ResourceResponseWriter) Header() http.Header {
m := rw.HeaderMap
if m == nil {
m = make(http.Header)
rw.HeaderMap = m
}
return m
}
// writeHeader writes a header if it was not written yet and
// detects Content-Type if needed.
//
// bytes or str are the beginning of the response body.
// We pass both to avoid unnecessarily generate garbage
// in rw.WriteString which was created for performance reasons.
// Non-nil bytes win.
func (rw *ResourceResponseWriter) writeHeader(b []byte, str string) {
if rw.wroteHeader {
return
}
if len(str) > 512 {
str = str[:512]
}
m := rw.Header()
_, hasType := m["Content-Type"]
hasTE := m.Get("Transfer-Encoding") != ""
if !hasType && !hasTE {
if b == nil {
b = []byte(str)
}
m.Set("Content-Type", http.DetectContentType(b))
}
rw.WriteHeader(200)
}
// Write implements http.ResponseWriter. The data in buf is written to
// rw.Body, if not nil.
func (rw *ResourceResponseWriter) Write(buf []byte) (int, error) {
rw.writeHeader(buf, "")
if rw.Body != nil {
rw.Body.Write(buf)
}
return len(buf), nil
}
// WriteHeader implements http.ResponseWriter.
func (rw *ResourceResponseWriter) WriteHeader(code int) {
if rw.wroteHeader {
return
}
rw.Code = code
rw.wroteHeader = true
if rw.HeaderMap == nil {
rw.HeaderMap = make(http.Header)
}
}
// Flush implements http.Flusher.
func (rw *ResourceResponseWriter) Flush() {
if !rw.wroteHeader {
rw.WriteHeader(200)
}
}
// Result returns the response generated by the handler.
func (rw *ResourceResponseWriter) Result() *pluginv2.CallResource_Response {
res := &pluginv2.CallResource_Response{
Code: int32(rw.Code),
Headers: map[string]*pluginv2.CallResource_StringList{},
}
if rw.Body != nil {
res.Body = rw.Body.Bytes()
}
for key, values := range rw.Header().Clone() {
res.Headers[key] = &pluginv2.CallResource_StringList{
Values: values,
}
}
return res
}