IntenseWebs/grafana
3
0
Fork 0
mirror of https://github.com/grafana/grafana.git synced 2025-02-25 18:55:37 -06:00
Code Issues Packages Projects Releases Wiki Activity

Plugins: Expose PID through proto interface (#77821)

Browse Source 
* expose PID thru interface

* apply PR feedback

* simplify

* add read lock
This commit is contained in:
Will Browne 2023-11-13 14:41:53 +01:00 committed by GitHub
parent edde17741d
commit 4299efbc56
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 47 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
pkg/plugins/backendplugin/grpcplugin/client_proto.go
View File

@ -3,6 +3,7 @@ package grpcplugin
import ( import (
"context" "context"
"errors" "errors"
"sync"
"google.golang.org/grpc" "google.golang.org/grpc"
@ -23,6 +24,7 @@ type ProtoClient interface {
pluginv2.DiagnosticsClient pluginv2.DiagnosticsClient
pluginv2.StreamClient pluginv2.StreamClient
PID() (string, error)
PluginID() string PluginID() string
Logger() log.Logger Logger() log.Logger
Start(context.Context) error Start(context.Context) error
@ -31,6 +33,8 @@ type ProtoClient interface {
type protoClient struct { type protoClient struct {
plugin *grpcPlugin plugin *grpcPlugin
mu sync.RWMutex
} }
type ProtoClientOpts struct { type ProtoClientOpts struct {
@ -57,6 +61,13 @@ func NewProtoClient(opts ProtoClientOpts) (ProtoClient, error) {
return &protoClient{plugin: p}, nil return &protoClient{plugin: p}, nil
} }
func (r *protoClient) PID() (string, error) {
if _, exists := r.client(); !exists {
return "", errClientNotStarted
}
return r.plugin.client.ID(), nil
}
func (r *protoClient) PluginID() string { func (r *protoClient) PluginID() string {
return r.plugin.descriptor.pluginID return r.plugin.descriptor.pluginID
} }
@ -66,58 +77,80 @@ func (r *protoClient) Logger() log.Logger {
} }
func (r *protoClient) Start(ctx context.Context) error { func (r *protoClient) Start(ctx context.Context) error {
r.mu.Lock()
defer r.mu.Unlock()
return r.plugin.Start(ctx) return r.plugin.Start(ctx)
} }
func (r *protoClient) Stop(ctx context.Context) error { func (r *protoClient) Stop(ctx context.Context) error {
r.mu.Lock()
defer r.mu.Unlock()
return r.plugin.Stop(ctx) return r.plugin.Stop(ctx)
} }
func (r *protoClient) QueryData(ctx context.Context, in *pluginv2.QueryDataRequest, opts ...grpc.CallOption) (*pluginv2.QueryDataResponse, error) { func (r *protoClient) client() (*ClientV2, bool) {
r.mu.RLock()
if r.plugin.pluginClient == nil { if r.plugin.pluginClient == nil {
r.mu.RUnlock()
return nil, false
}
pc := r.plugin.pluginClient
r.mu.RUnlock()
return pc, true
}
func (r *protoClient) QueryData(ctx context.Context, in *pluginv2.QueryDataRequest, opts ...grpc.CallOption) (*pluginv2.QueryDataResponse, error) {
c, exists := r.client()
if !exists {
return nil, errClientNotStarted return nil, errClientNotStarted
} }
return r.plugin.pluginClient.DataClient.QueryData(ctx, in, opts...) return c.DataClient.QueryData(ctx, in, opts...)
} }
func (r *protoClient) CallResource(ctx context.Context, in *pluginv2.CallResourceRequest, opts ...grpc.CallOption) (pluginv2.Resource_CallResourceClient, error) { func (r *protoClient) CallResource(ctx context.Context, in *pluginv2.CallResourceRequest, opts ...grpc.CallOption) (pluginv2.Resource_CallResourceClient, error) {
if r.plugin.pluginClient == nil { c, exists := r.client()
if !exists {
return nil, errClientNotStarted return nil, errClientNotStarted
} }
return r.plugin.pluginClient.ResourceClient.CallResource(ctx, in, opts...) return c.ResourceClient.CallResource(ctx, in, opts...)
} }
func (r *protoClient) CheckHealth(ctx context.Context, in *pluginv2.CheckHealthRequest, opts ...grpc.CallOption) (*pluginv2.CheckHealthResponse, error) { func (r *protoClient) CheckHealth(ctx context.Context, in *pluginv2.CheckHealthRequest, opts ...grpc.CallOption) (*pluginv2.CheckHealthResponse, error) {
if r.plugin.pluginClient == nil { c, exists := r.client()
if !exists {
return nil, errClientNotStarted return nil, errClientNotStarted
} }
return r.plugin.pluginClient.DiagnosticsClient.CheckHealth(ctx, in, opts...) return c.DiagnosticsClient.CheckHealth(ctx, in, opts...)
} }
func (r *protoClient) CollectMetrics(ctx context.Context, in *pluginv2.CollectMetricsRequest, opts ...grpc.CallOption) (*pluginv2.CollectMetricsResponse, error) { func (r *protoClient) CollectMetrics(ctx context.Context, in *pluginv2.CollectMetricsRequest, opts ...grpc.CallOption) (*pluginv2.CollectMetricsResponse, error) {
if r.plugin.pluginClient == nil { c, exists := r.client()
if !exists {
return nil, errClientNotStarted return nil, errClientNotStarted
} }
return r.plugin.pluginClient.DiagnosticsClient.CollectMetrics(ctx, in, opts...) return c.DiagnosticsClient.CollectMetrics(ctx, in, opts...)
} }
func (r *protoClient) SubscribeStream(ctx context.Context, in *pluginv2.SubscribeStreamRequest, opts ...grpc.CallOption) (*pluginv2.SubscribeStreamResponse, error) { func (r *protoClient) SubscribeStream(ctx context.Context, in *pluginv2.SubscribeStreamRequest, opts ...grpc.CallOption) (*pluginv2.SubscribeStreamResponse, error) {
if r.plugin.pluginClient == nil { c, exists := r.client()
if !exists {
return nil, errClientNotStarted return nil, errClientNotStarted
} }
return r.plugin.pluginClient.StreamClient.SubscribeStream(ctx, in, opts...) return c.StreamClient.SubscribeStream(ctx, in, opts...)
} }
func (r *protoClient) RunStream(ctx context.Context, in *pluginv2.RunStreamRequest, opts ...grpc.CallOption) (pluginv2.Stream_RunStreamClient, error) { func (r *protoClient) RunStream(ctx context.Context, in *pluginv2.RunStreamRequest, opts ...grpc.CallOption) (pluginv2.Stream_RunStreamClient, error) {
if r.plugin.pluginClient == nil { c, exists := r.client()
if !exists {
return nil, errClientNotStarted return nil, errClientNotStarted
} }
return r.plugin.pluginClient.StreamClient.RunStream(ctx, in, opts...) return c.StreamClient.RunStream(ctx, in, opts...)
} }
func (r *protoClient) PublishStream(ctx context.Context, in *pluginv2.PublishStreamRequest, opts ...grpc.CallOption) (*pluginv2.PublishStreamResponse, error) { func (r *protoClient) PublishStream(ctx context.Context, in *pluginv2.PublishStreamRequest, opts ...grpc.CallOption) (*pluginv2.PublishStreamResponse, error) {
if r.plugin.pluginClient == nil { c, exists := r.client()
if !exists {
return nil, errClientNotStarted return nil, errClientNotStarted
} }
return r.plugin.pluginClient.StreamClient.PublishStream(ctx, in, opts...) return c.StreamClient.PublishStream(ctx, in, opts...)
} }