Plugins: Add running method to interface (#78326)

* add running to interface

* add context
This commit is contained in:
Will Browne
2023-11-17 13:52:31 +01:00
committed by GitHub
parent a9208f623a
commit ddb7406caa
@@ -24,12 +24,13 @@ type ProtoClient interface {
pluginv2.DiagnosticsClient pluginv2.DiagnosticsClient
pluginv2.StreamClient pluginv2.StreamClient
PID() (string, error) PID(context.Context) (string, error)
PluginID() string PluginID() string
PluginVersion() string PluginVersion() string
Logger() log.Logger Logger() log.Logger
Start(context.Context) error Start(context.Context) error
Stop(context.Context) error Stop(context.Context) error
Running(context.Context) bool
} }
type protoClient struct { type protoClient struct {
@@ -64,8 +65,8 @@ func NewProtoClient(opts ProtoClientOpts) (ProtoClient, error) {
return &protoClient{plugin: p, pluginVersion: opts.PluginVersion}, nil return &protoClient{plugin: p, pluginVersion: opts.PluginVersion}, nil
} }
func (r *protoClient) PID() (string, error) { func (r *protoClient) PID(ctx context.Context) (string, error) {
if _, exists := r.client(); !exists { if _, exists := r.client(ctx); !exists {
return "", errClientNotStarted return "", errClientNotStarted
} }
return r.plugin.client.ID(), nil return r.plugin.client.ID(), nil
@@ -95,7 +96,17 @@ func (r *protoClient) Stop(ctx context.Context) error {
return r.plugin.Stop(ctx) return r.plugin.Stop(ctx)
} }
func (r *protoClient) client() (*ClientV2, bool) { func (r *protoClient) Running(_ context.Context) bool {
r.mu.RLock()
defer r.mu.RUnlock()
return !r.plugin.Exited()
}
func (r *protoClient) client(ctx context.Context) (*ClientV2, bool) {
if !r.Running(ctx) {
return nil, false
}
r.mu.RLock() r.mu.RLock()
if r.plugin.pluginClient == nil { if r.plugin.pluginClient == nil {
r.mu.RUnlock() r.mu.RUnlock()
@@ -107,7 +118,7 @@ func (r *protoClient) client() (*ClientV2, bool) {
} }
func (r *protoClient) QueryData(ctx context.Context, in *pluginv2.QueryDataRequest, opts ...grpc.CallOption) (*pluginv2.QueryDataResponse, error) { func (r *protoClient) QueryData(ctx context.Context, in *pluginv2.QueryDataRequest, opts ...grpc.CallOption) (*pluginv2.QueryDataResponse, error) {
c, exists := r.client() c, exists := r.client(ctx)
if !exists { if !exists {
return nil, errClientNotStarted return nil, errClientNotStarted
} }
@@ -115,7 +126,7 @@ func (r *protoClient) QueryData(ctx context.Context, in *pluginv2.QueryDataReque
} }
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) {
c, exists := r.client() c, exists := r.client(ctx)
if !exists { if !exists {
return nil, errClientNotStarted return nil, errClientNotStarted
} }
@@ -123,7 +134,7 @@ func (r *protoClient) CallResource(ctx context.Context, in *pluginv2.CallResourc
} }
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) {
c, exists := r.client() c, exists := r.client(ctx)
if !exists { if !exists {
return nil, errClientNotStarted return nil, errClientNotStarted
} }
@@ -131,7 +142,7 @@ func (r *protoClient) CheckHealth(ctx context.Context, in *pluginv2.CheckHealthR
} }
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) {
c, exists := r.client() c, exists := r.client(ctx)
if !exists { if !exists {
return nil, errClientNotStarted return nil, errClientNotStarted
} }
@@ -139,7 +150,7 @@ func (r *protoClient) CollectMetrics(ctx context.Context, in *pluginv2.CollectMe
} }
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) {
c, exists := r.client() c, exists := r.client(ctx)
if !exists { if !exists {
return nil, errClientNotStarted return nil, errClientNotStarted
} }
@@ -147,7 +158,7 @@ func (r *protoClient) SubscribeStream(ctx context.Context, in *pluginv2.Subscrib
} }
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) {
c, exists := r.client() c, exists := r.client(ctx)
if !exists { if !exists {
return nil, errClientNotStarted return nil, errClientNotStarted
} }
@@ -155,7 +166,7 @@ func (r *protoClient) RunStream(ctx context.Context, in *pluginv2.RunStreamReque
} }
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) {
c, exists := r.client() c, exists := r.client(ctx)
if !exists { if !exists {
return nil, errClientNotStarted return nil, errClientNotStarted
} }