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

Live: proxy subscribe data for subscribe and runstream plugin calls (#42824)

Browse Source 
Co-authored-by: Ryan McKinley <ryantxu@gmail.com>
This commit is contained in:
Alexander Emelin
2021-12-14 20:12:00 +03:00
committed by GitHub
parent d2c5dac76f
commit f4cc353225
15 changed files with 62 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
pkg/models/live.go
View File

@@ -18,6 +18,7 @@ type ChannelClientCount func(orgID int64, channel string) (int, error)
type SubscribeEvent struct {
Channel string
Path string
Data json.RawMessage
}
// SubscribeReply is a reaction to SubscribeEvent.

3
pkg/services/live/features/plugin.go
View File

@@ -73,6 +73,7 @@ func (r *PluginPathRunner) OnSubscribe(ctx context.Context, user *models.SignedI
resp, err := r.handler.SubscribeStream(ctx, &backend.SubscribeStreamRequest{
PluginContext: pCtx,
Path: r.path,
Data: e.Data,
})
if err != nil {
logger.Error("Plugin OnSubscribe call error", "error", err, "path", r.path)
@@ -82,7 +83,7 @@ func (r *PluginPathRunner) OnSubscribe(ctx context.Context, user *models.SignedI
return models.SubscribeReply{}, resp.Status, nil
}
submitResult, err := r.runStreamManager.SubmitStream(ctx, user, orgchannel.PrependOrgID(user.OrgId, e.Channel), r.path, pCtx, r.handler, false)
submitResult, err := r.runStreamManager.SubmitStream(ctx, user, orgchannel.PrependOrgID(user.OrgId, e.Channel), r.path, e.Data, pCtx, r.handler, false)
if err != nil {
logger.Error("Error submitting stream to manager", "error", err, "path", r.path)
return models.SubscribeReply{}, 0, centrifuge.ErrorInternal

3
pkg/services/live/live.go
View File

@@ -597,7 +597,7 @@ func (g *GrafanaLive) handleOnSubscribe(client *centrifuge.Client, e centrifuge.
reply, status, err = sub.Subscribe(client.Context(), pipeline.Vars{
OrgID: orgID,
Channel: channel,
})
}, e.Data)
if err != nil {
logger.Error("Error channel rule subscribe", "user", client.UserID(), "client", client.ID(), "channel", e.Channel, "error", err)
return centrifuge.SubscribeReply{}, centrifuge.ErrorInternal
@@ -622,6 +622,7 @@ func (g *GrafanaLive) handleOnSubscribe(client *centrifuge.Client, e centrifuge.
reply, status, err = handler.OnSubscribe(client.Context(), user, models.SubscribeEvent{
Channel: channel,
Path: addr.Path,
Data: e.Data,
})
if err != nil {
logger.Error("Error calling channel handler subscribe", "user", client.UserID(), "client", client.ID(), "channel", e.Channel, "error", err)

2
pkg/services/live/pipeline/pipeline.go
View File

@@ -107,7 +107,7 @@ type FrameOutputter interface {
// Subscriber can handle channel subscribe events.
type Subscriber interface {
Type() string
Subscribe(ctx context.Context, vars Vars) (models.SubscribeReply, backend.SubscribeStreamStatus, error)
Subscribe(ctx context.Context, vars Vars, data []byte) (models.SubscribeReply, backend.SubscribeStreamStatus, error)
}
// PublishAuthChecker checks whether current user can publish to a channel.

3
pkg/services/live/pipeline/subscribe_builtin.go
View File

@@ -28,7 +28,7 @@ func (s *BuiltinSubscriber) Type() string {
return SubscriberTypeBuiltin
}
func (s *BuiltinSubscriber) Subscribe(ctx context.Context, vars Vars) (models.SubscribeReply, backend.SubscribeStreamStatus, error) {
func (s *BuiltinSubscriber) Subscribe(ctx context.Context, vars Vars, data []byte) (models.SubscribeReply, backend.SubscribeStreamStatus, error) {
u, ok := livecontext.GetContextSignedUser(ctx)
if !ok {
return models.SubscribeReply{}, backend.SubscribeStreamStatusPermissionDenied, nil
@@ -40,5 +40,6 @@ func (s *BuiltinSubscriber) Subscribe(ctx context.Context, vars Vars) (models.Su
return handler.OnSubscribe(ctx, u, models.SubscribeEvent{
Channel: vars.Channel,
Path: vars.Path,
Data: data,
})
}

2
pkg/services/live/pipeline/subscribe_managed_stream.go
View File

@@ -24,7 +24,7 @@ func (s *ManagedStreamSubscriber) Type() string {
return SubscriberTypeManagedStream
}
func (s *ManagedStreamSubscriber) Subscribe(ctx context.Context, vars Vars) (models.SubscribeReply, backend.SubscribeStreamStatus, error) {
func (s *ManagedStreamSubscriber) Subscribe(ctx context.Context, vars Vars, _ []byte) (models.SubscribeReply, backend.SubscribeStreamStatus, error) {
stream, err := s.managedStream.GetOrCreateStream(vars.OrgID, vars.Scope, vars.Namespace)
if err != nil {
logger.Error("Error getting managed stream", "error", err)

4
pkg/services/live/pipeline/subscribe_multiple.go
View File

@@ -21,11 +21,11 @@ func (s *MultipleSubscriber) Type() string {
return SubscriberTypeMultiple
}
func (s *MultipleSubscriber) Subscribe(ctx context.Context, vars Vars) (models.SubscribeReply, backend.SubscribeStreamStatus, error) {
func (s *MultipleSubscriber) Subscribe(ctx context.Context, vars Vars, data []byte) (models.SubscribeReply, backend.SubscribeStreamStatus, error) {
finalReply := models.SubscribeReply{}
for _, s := range s.Subscribers {
reply, status, err := s.Subscribe(ctx, vars)
reply, status, err := s.Subscribe(ctx, vars, data)
if err != nil {
return models.SubscribeReply{}, 0, err
}

7
pkg/services/live/runstream/manager.go
View File

@@ -136,7 +136,7 @@ func (s *Manager) handleDatasourceEvent(orgID int64, dsUID string, resubmit bool
if resubmit {
// Re-submit streams.
for _, sr := range resubmitRequests {
_, err := s.SubmitStream(s.baseCtx, sr.user, sr.Channel, sr.Path, sr.PluginContext, sr.StreamRunner, true)
_, err := s.SubmitStream(s.baseCtx, sr.user, sr.Channel, sr.Path, sr.Data, sr.PluginContext, sr.StreamRunner, true)
if err != nil {
// Log error but do not prevent execution of caller routine.
logger.Error("Error re-submitting stream", "path", sr.Path, "error", err)
@@ -301,6 +301,7 @@ func (s *Manager) runStream(ctx context.Context, cancelFn func(), sr streamReque
&backend.RunStreamRequest{
PluginContext: pluginCtx,
Path: sr.Path,
Data: sr.Data,
},
backend.NewStreamSender(&packetSender{channelLocalPublisher: s.channelSender, channel: sr.Channel}),
)
@@ -375,6 +376,7 @@ type streamRequest struct {
user *models.SignedInUser
PluginContext backend.PluginContext
StreamRunner StreamRunner
Data []byte
}
type submitRequest struct {
@@ -398,7 +400,7 @@ var errDatasourceNotFound = errors.New("datasource not found")
// SubmitStream submits stream handler in Manager to manage.
// The stream will be opened and kept till channel has active subscribers.
func (s *Manager) SubmitStream(ctx context.Context, user *models.SignedInUser, channel string, path string, pCtx backend.PluginContext, streamRunner StreamRunner, isResubmit bool) (*submitResult, error) {
func (s *Manager) SubmitStream(ctx context.Context, user *models.SignedInUser, channel string, path string, data []byte, pCtx backend.PluginContext, streamRunner StreamRunner, isResubmit bool) (*submitResult, error) {
if isResubmit {
// Resolve new plugin context as it could be modified since last call.
var datasourceUID string
@@ -423,6 +425,7 @@ func (s *Manager) SubmitStream(ctx context.Context, user *models.SignedInUser, c
Path: path,
PluginContext: pCtx,
StreamRunner: streamRunner,
Data: data,
},
}

18
pkg/services/live/runstream/manager_test.go
View File

@@ -94,12 +94,12 @@ func TestStreamManager_SubmitStream_Send(t *testing.T) {
return ctx.Err()
}).Times(1)
result, err := manager.SubmitStream(context.Background(), &models.SignedInUser{UserId: 2, OrgId: 1}, "1/test", "test", testPluginContext, mockStreamRunner, false)
result, err := manager.SubmitStream(context.Background(), &models.SignedInUser{UserId: 2, OrgId: 1}, "1/test", "test", nil, testPluginContext, mockStreamRunner, false)
require.NoError(t, err)
require.False(t, result.StreamExists)
// try submit the same.
result, err = manager.SubmitStream(context.Background(), &models.SignedInUser{UserId: 2, OrgId: 1}, "1/test", "test", backend.PluginContext{}, mockStreamRunner, false)
result, err = manager.SubmitStream(context.Background(), &models.SignedInUser{UserId: 2, OrgId: 1}, "1/test", "test", nil, backend.PluginContext{}, mockStreamRunner, false)
require.NoError(t, err)
require.True(t, result.StreamExists)
@@ -163,12 +163,12 @@ func TestStreamManager_SubmitStream_DifferentOrgID(t *testing.T) {
return ctx.Err()
}).Times(1)
result, err := manager.SubmitStream(context.Background(), &models.SignedInUser{UserId: 2, OrgId: 1}, "1/test", "test", backend.PluginContext{}, mockStreamRunner1, false)
result, err := manager.SubmitStream(context.Background(), &models.SignedInUser{UserId: 2, OrgId: 1}, "1/test", "test", nil, backend.PluginContext{}, mockStreamRunner1, false)
require.NoError(t, err)
require.False(t, result.StreamExists)
// try submit the same channel but different orgID.
result, err = manager.SubmitStream(context.Background(), &models.SignedInUser{UserId: 2, OrgId: 2}, "2/test", "test", backend.PluginContext{}, mockStreamRunner2, false)
result, err = manager.SubmitStream(context.Background(), &models.SignedInUser{UserId: 2, OrgId: 2}, "2/test", "test", nil, backend.PluginContext{}, mockStreamRunner2, false)
require.NoError(t, err)
require.False(t, result.StreamExists)
@@ -219,7 +219,7 @@ func TestStreamManager_SubmitStream_CloseNoSubscribers(t *testing.T) {
return ctx.Err()
}).Times(1)
_, err := manager.SubmitStream(context.Background(), &models.SignedInUser{UserId: 2, OrgId: 1}, "1/test", "test", backend.PluginContext{}, mockStreamRunner, false)
_, err := manager.SubmitStream(context.Background(), &models.SignedInUser{UserId: 2, OrgId: 1}, "1/test", "test", nil, backend.PluginContext{}, mockStreamRunner, false)
require.NoError(t, err)
waitWithTimeout(t, startedCh, time.Second)
@@ -273,7 +273,7 @@ func TestStreamManager_SubmitStream_ErrorRestartsRunStream(t *testing.T) {
return errors.New("boom")
}).Times(numErrors + 1)
result, err := manager.SubmitStream(context.Background(), &models.SignedInUser{UserId: 2, OrgId: 1}, "test", "test", testPluginContext, mockStreamRunner, false)
result, err := manager.SubmitStream(context.Background(), &models.SignedInUser{UserId: 2, OrgId: 1}, "test", "test", nil, testPluginContext, mockStreamRunner, false)
require.NoError(t, err)
require.False(t, result.StreamExists)
@@ -307,7 +307,7 @@ func TestStreamManager_SubmitStream_NilErrorStopsRunStream(t *testing.T) {
return nil
}).Times(1)
result, err := manager.SubmitStream(context.Background(), &models.SignedInUser{UserId: 2, OrgId: 1}, "test", "test", backend.PluginContext{}, mockStreamRunner, false)
result, err := manager.SubmitStream(context.Background(), &models.SignedInUser{UserId: 2, OrgId: 1}, "test", "test", nil, backend.PluginContext{}, mockStreamRunner, false)
require.NoError(t, err)
require.False(t, result.StreamExists)
waitWithTimeout(t, result.CloseNotify, time.Second)
@@ -366,7 +366,7 @@ func TestStreamManager_HandleDatasourceUpdate(t *testing.T) {
return nil
}).Times(2)
result, err := manager.SubmitStream(context.Background(), &models.SignedInUser{UserId: 2, OrgId: 1}, "test", "test", testPluginContext, mockStreamRunner, false)
result, err := manager.SubmitStream(context.Background(), &models.SignedInUser{UserId: 2, OrgId: 1}, "test", "test", nil, testPluginContext, mockStreamRunner, false)
require.NoError(t, err)
require.False(t, result.StreamExists)
@@ -422,7 +422,7 @@ func TestStreamManager_HandleDatasourceDelete(t *testing.T) {
return ctx.Err()
}).Times(1)
result, err := manager.SubmitStream(context.Background(), &models.SignedInUser{UserId: 2, OrgId: 1}, "test", "test", testPluginContext, mockStreamRunner, false)
result, err := manager.SubmitStream(context.Background(), &models.SignedInUser{UserId: 2, OrgId: 1}, "test", "test", nil, testPluginContext, mockStreamRunner, false)
require.NoError(t, err)
require.False(t, result.StreamExists)