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

Backend Plugins: Support handling of streaming resource response (#22580)

Browse Source 
Use v0.19.0 of SDK.
Support handling of streaming resource response.
Disable gzip/compression middleware for resources 
to allow chunked/streaming response to clients the gzip
middleware had to be disabled since it buffers the full
response before sending it to the client.

Closes #22569

Co-Authored-By: Arve Knudsen <arve.knudsen@gmail.com>
This commit is contained in:
Marcus Efraimsson
2020-03-06 01:44:07 +07:00
committed by GitHub
co-authored by Arve Knudsen
parent f95c8b785c
commit 4ff613a432
17 changed files with 518 additions and 241 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/plugins/backendplugin/backend_plugin.go
+15 -16
View File
@@ -72,7 +72,7 @@ func (p *BackendPlugin) start(ctx context.Context) error {
if rawBackend != nil {
if plugin, ok := rawBackend.(CorePlugin); ok {
p.core = plugin
client.DatasourcePlugin = plugin
client.CorePlugin = plugin
}
}
@@ -186,8 +186,8 @@ func (p *BackendPlugin) checkHealth(ctx context.Context) (*pluginv2.CheckHealth_
if st, ok := status.FromError(err); ok {
if st.Code() == codes.Unimplemented {
return &pluginv2.CheckHealth_Response{
Status: pluginv2.CheckHealth_Response_UNKNOWN,
Info: "Health check not implemented",
Status: pluginv2.CheckHealth_Response_UNKNOWN,
Message: "Health check not implemented",
}, nil
}
}
@@ -197,9 +197,13 @@ func (p *BackendPlugin) checkHealth(ctx context.Context) (*pluginv2.CheckHealth_
return res, nil
}
func (p *BackendPlugin) callResource(ctx context.Context, req CallResourceRequest) (*CallResourceResult, error) {
func (p *BackendPlugin) callResource(ctx context.Context, req CallResourceRequest) (callResourceResultStream, error) {
p.logger.Debug("Calling resource", "path", req.Path, "method", req.Method)
if p.core == nil || p.client == nil || p.client.Exited() {
return nil, errors.New("plugin not running, cannot call resource")
}
reqHeaders := map[string]*pluginv2.CallResource_StringList{}
for k, v := range req.Headers {
reqHeaders[k] = &pluginv2.CallResource_StringList{Values: v}
@@ -238,12 +242,14 @@ func (p *BackendPlugin) callResource(ctx context.Context, req CallResourceReques
}
}
protoResp, err := p.core.CallResource(ctx, protoReq)
protoStream, 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,
return &singleCallResourceResult{
result: &CallResourceResult{
Status: http.StatusNotImplemented,
},
}, nil
}
}
@@ -251,15 +257,8 @@ func (p *BackendPlugin) callResource(ctx context.Context, req CallResourceReques
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),
return &callResourceResultStreamImpl{
stream: protoStream,
}, nil
}
pkg/plugins/backendplugin/client.go
+4 -10
View File
@@ -102,17 +102,11 @@ 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)
plugin.DiagnosticsServer
}
type CorePlugin interface {
CallResource(ctx context.Context, req *pluginv2.CallResource_Request) (*pluginv2.CallResource_Response, error)
DatasourcePlugin
plugin.CoreClient
}
type TransformPlugin interface {
@@ -127,6 +121,6 @@ type LegacyClient struct {
// Client client for communicating with a plugin using the current plugin protocol.
type Client struct {
DatasourcePlugin DatasourcePlugin
TransformPlugin TransformPlugin
CorePlugin CorePlugin
TransformPlugin TransformPlugin
}
pkg/plugins/backendplugin/contracts.go
+47 -4
View File
@@ -37,8 +37,8 @@ func (hs HealthStatus) String() string {
// CheckHealthResult check health result.
type CheckHealthResult struct {
Status HealthStatus
Info string
Status HealthStatus
Message string
}
func checkHealthResultFromProto(protoResp *pluginv2.CheckHealth_Response) *CheckHealthResult {
@@ -51,8 +51,8 @@ func checkHealthResultFromProto(protoResp *pluginv2.CheckHealth_Response) *Check
}
return &CheckHealthResult{
Status: status,
Info: protoResp.Info,
Status: status,
Message: protoResp.Message,
}
}
@@ -91,3 +91,46 @@ type CallResourceResult struct {
Headers map[string][]string
Body []byte
}
type callResourceResultStream interface {
Recv() (*CallResourceResult, error)
Close() error
}
type callResourceResultStreamImpl struct {
stream pluginv2.Core_CallResourceClient
}
func (s *callResourceResultStreamImpl) Recv() (*CallResourceResult, error) {
protoResp, err := s.stream.Recv()
if err != nil {
return nil, 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
}
func (s *callResourceResultStreamImpl) Close() error {
return s.stream.CloseSend()
}
type singleCallResourceResult struct {
result *CallResourceResult
}
func (s *singleCallResourceResult) Recv() (*CallResourceResult, error) {
return s.result, nil
}
func (s *singleCallResourceResult) Close() error {
return nil
}
pkg/plugins/backendplugin/manager.go
+45 -15
View File
@@ -3,6 +3,7 @@ package backendplugin
import (
"context"
"errors"
"io"
"sync"
"time"
@@ -209,30 +210,59 @@ func (m *manager) CallResource(config PluginConfig, c *models.ReqContext, path s
Body: body,
}
res, err := p.callResource(clonedReq.Context(), req)
stream, err := p.callResource(clonedReq.Context(), req)
if err != nil {
c.JsonApiErr(500, "Failed to call resource", err)
return
}
// Make sure a content type always is returned in response
if _, exists := res.Headers["Content-Type"]; !exists {
res.Headers["Content-Type"] = []string{"application/json"}
}
processedStreams := 0
for k, values := range res.Headers {
if k == "Set-Cookie" {
continue
for {
resp, err := stream.Recv()
if err == io.EOF {
if processedStreams == 0 {
c.JsonApiErr(500, "Received empty resource response ", nil)
}
return
}
if err != nil {
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
}
for _, v := range values {
c.Resp.Header().Add(k, v)
}
}
// Expected that headers and status are only part of first stream
if processedStreams == 0 {
// Make sure a content type always is returned in response
if _, exists := resp.Headers["Content-Type"]; !exists {
resp.Headers["Content-Type"] = []string{"application/json"}
}
c.WriteHeader(res.Status)
if _, err := c.Write(res.Body); err != nil {
p.logger.Error("Failed to write resource response", "error", err)
for k, values := range resp.Headers {
// Due to security reasons we don't want to forward
// cookies from a backend plugin to clients/browsers.
if k == "Set-Cookie" {
continue
}
for _, v := range values {
c.Resp.Header().Add(k, v)
}
}
c.WriteHeader(resp.Status)
}
if _, err := c.Write(resp.Body); err != nil {
p.logger.Error("Failed to write resource response", "error", err)
}
c.Resp.Flush()
processedStreams++
}
}