Basic streaming plugin support (#31940)

This pull request migrates testdata to coreplugin streaming capabilities,
this is mostly a working concept of streaming plugins at the moment, 
the work will continue in the following pull requests.

pkg/services/live/channel.go

@@ -6,13 +6,14 @@ import (
 
 // ChannelAddress is the channel ID split by parts.
 type ChannelAddress struct {
-	// Scope is "grafana", "ds", or "plugin".
+	// Scope is one of available channel scopes:
+	// like ScopeGrafana, ScopePlugin, ScopeDatasource.
 	Scope string `json:"scope,omitempty"`
 
 	// Namespace meaning depends on the scope.
-	// * when grafana, namespace is a "feature"
-	// * when ds, namespace is the datasource id
-	// * when plugin, namespace is the plugin name
+	// * when ScopeGrafana, namespace is a "feature"
+	// * when ScopePlugin, namespace is the plugin name
+	// * when ScopeDatasource, namespace is the datasource uid
 	Namespace string `json:"namespace,omitempty"`
 
 	// Within each namespace, the handler can process the path as needed.

pkg/services/live/context.go

@@ -0,0 +1,24 @@
+package live
+
+import (
+	"context"
+
+	"github.com/grafana/grafana/pkg/models"
+)
+
+type signedUserContextKeyType int
+
+var signedUserContextKey signedUserContextKeyType
+
+func setContextSignedUser(ctx context.Context, user *models.SignedInUser) context.Context {
+	ctx = context.WithValue(ctx, signedUserContextKey, user)
+	return ctx
+}
+
+func getContextSignedUser(ctx context.Context) (*models.SignedInUser, bool) {
+	if val := ctx.Value(signedUserContextKey); val != nil {
+		user, ok := val.(*models.SignedInUser)
+		return user, ok
+	}
+	return nil, false
+}

pkg/services/live/features/mock.go

@@ -0,0 +1,164 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: github.com/grafana/grafana/pkg/services/live/features (interfaces: ChannelPublisher,PresenceGetter,PluginContextGetter,StreamRunner)
+
+// Package features is a generated GoMock package.
+package features
+
+import (
+	context "context"
+	reflect "reflect"
+
+	gomock "github.com/golang/mock/gomock"
+	backend "github.com/grafana/grafana-plugin-sdk-go/backend"
+)
+
+// MockChannelPublisher is a mock of ChannelPublisher interface.
+type MockChannelPublisher struct {
+	ctrl     *gomock.Controller
+	recorder *MockChannelPublisherMockRecorder
+}
+
+// MockChannelPublisherMockRecorder is the mock recorder for MockChannelPublisher.
+type MockChannelPublisherMockRecorder struct {
+	mock *MockChannelPublisher
+}
+
+// NewMockChannelPublisher creates a new mock instance.
+func NewMockChannelPublisher(ctrl *gomock.Controller) *MockChannelPublisher {
+	mock := &MockChannelPublisher{ctrl: ctrl}
+	mock.recorder = &MockChannelPublisherMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockChannelPublisher) EXPECT() *MockChannelPublisherMockRecorder {
+	return m.recorder
+}
+
+// Publish mocks base method.
+func (m *MockChannelPublisher) Publish(arg0 string, arg1 []byte) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Publish", arg0, arg1)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Publish indicates an expected call of Publish.
+func (mr *MockChannelPublisherMockRecorder) Publish(arg0, arg1 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Publish", reflect.TypeOf((*MockChannelPublisher)(nil).Publish), arg0, arg1)
+}
+
+// MockPresenceGetter is a mock of PresenceGetter interface.
+type MockPresenceGetter struct {
+	ctrl     *gomock.Controller
+	recorder *MockPresenceGetterMockRecorder
+}
+
+// MockPresenceGetterMockRecorder is the mock recorder for MockPresenceGetter.
+type MockPresenceGetterMockRecorder struct {
+	mock *MockPresenceGetter
+}
+
+// NewMockPresenceGetter creates a new mock instance.
+func NewMockPresenceGetter(ctrl *gomock.Controller) *MockPresenceGetter {
+	mock := &MockPresenceGetter{ctrl: ctrl}
+	mock.recorder = &MockPresenceGetterMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockPresenceGetter) EXPECT() *MockPresenceGetterMockRecorder {
+	return m.recorder
+}
+
+// GetNumSubscribers mocks base method.
+func (m *MockPresenceGetter) GetNumSubscribers(arg0 string) (int, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetNumSubscribers", arg0)
+	ret0, _ := ret[0].(int)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetNumSubscribers indicates an expected call of GetNumSubscribers.
+func (mr *MockPresenceGetterMockRecorder) GetNumSubscribers(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetNumSubscribers", reflect.TypeOf((*MockPresenceGetter)(nil).GetNumSubscribers), arg0)
+}
+
+// MockPluginContextGetter is a mock of PluginContextGetter interface.
+type MockPluginContextGetter struct {
+	ctrl     *gomock.Controller
+	recorder *MockPluginContextGetterMockRecorder
+}
+
+// MockPluginContextGetterMockRecorder is the mock recorder for MockPluginContextGetter.
+type MockPluginContextGetterMockRecorder struct {
+	mock *MockPluginContextGetter
+}
+
+// NewMockPluginContextGetter creates a new mock instance.
+func NewMockPluginContextGetter(ctrl *gomock.Controller) *MockPluginContextGetter {
+	mock := &MockPluginContextGetter{ctrl: ctrl}
+	mock.recorder = &MockPluginContextGetterMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockPluginContextGetter) EXPECT() *MockPluginContextGetterMockRecorder {
+	return m.recorder
+}
+
+// GetPluginContext mocks base method.
+func (m *MockPluginContextGetter) GetPluginContext(arg0 context.Context, arg1, arg2 string) (backend.PluginContext, bool, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetPluginContext", arg0, arg1, arg2)
+	ret0, _ := ret[0].(backend.PluginContext)
+	ret1, _ := ret[1].(bool)
+	ret2, _ := ret[2].(error)
+	return ret0, ret1, ret2
+}
+
+// GetPluginContext indicates an expected call of GetPluginContext.
+func (mr *MockPluginContextGetterMockRecorder) GetPluginContext(arg0, arg1, arg2 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetPluginContext", reflect.TypeOf((*MockPluginContextGetter)(nil).GetPluginContext), arg0, arg1, arg2)
+}
+
+// MockStreamRunner is a mock of StreamRunner interface.
+type MockStreamRunner struct {
+	ctrl     *gomock.Controller
+	recorder *MockStreamRunnerMockRecorder
+}
+
+// MockStreamRunnerMockRecorder is the mock recorder for MockStreamRunner.
+type MockStreamRunnerMockRecorder struct {
+	mock *MockStreamRunner
+}
+
+// NewMockStreamRunner creates a new mock instance.
+func NewMockStreamRunner(ctrl *gomock.Controller) *MockStreamRunner {
+	mock := &MockStreamRunner{ctrl: ctrl}
+	mock.recorder = &MockStreamRunnerMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockStreamRunner) EXPECT() *MockStreamRunnerMockRecorder {
+	return m.recorder
+}
+
+// RunStream mocks base method.
+func (m *MockStreamRunner) RunStream(arg0 context.Context, arg1 *backend.RunStreamRequest, arg2 backend.StreamPacketSender) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "RunStream", arg0, arg1, arg2)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// RunStream indicates an expected call of RunStream.
+func (mr *MockStreamRunnerMockRecorder) RunStream(arg0, arg1, arg2 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RunStream", reflect.TypeOf((*MockStreamRunner)(nil).RunStream), arg0, arg1, arg2)
+}

pkg/services/live/features/plugin.go

@@ -0,0 +1,122 @@
+package features
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/centrifugal/centrifuge"
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"github.com/grafana/grafana/pkg/models"
+)
+
+//go:generate mockgen -destination=mock.go -package=features github.com/grafana/grafana/pkg/services/live/features ChannelPublisher,PresenceGetter,PluginContextGetter,StreamRunner
+
+type ChannelPublisher interface {
+	Publish(channel string, data []byte) error
+}
+
+type PresenceGetter interface {
+	GetNumSubscribers(channel string) (int, error)
+}
+
+type PluginContextGetter interface {
+	GetPluginContext(ctx context.Context, pluginID string, datasourceUID string) (backend.PluginContext, bool, error)
+}
+
+type StreamRunner interface {
+	RunStream(ctx context.Context, request *backend.RunStreamRequest, sender backend.StreamPacketSender) error
+}
+
+type streamSender struct {
+	channel          string
+	channelPublisher ChannelPublisher
+}
+
+func newStreamSender(channel string, publisher ChannelPublisher) *streamSender {
+	return &streamSender{channel: channel, channelPublisher: publisher}
+}
+
+func (p *streamSender) Send(packet *backend.StreamPacket) error {
+	return p.channelPublisher.Publish(p.channel, packet.Payload)
+}
+
+// PluginRunner can handle streaming operations for channels belonging to plugins.
+type PluginRunner struct {
+	pluginID            string
+	datasourceUID       string
+	pluginContextGetter PluginContextGetter
+	handler             backend.StreamHandler
+	streamManager       *StreamManager
+}
+
+// NewPluginRunner creates new PluginRunner.
+func NewPluginRunner(pluginID string, datasourceUID string, streamManager *StreamManager, pluginContextGetter PluginContextGetter, handler backend.StreamHandler) *PluginRunner {
+	return &PluginRunner{
+		pluginID:            pluginID,
+		datasourceUID:       datasourceUID,
+		pluginContextGetter: pluginContextGetter,
+		handler:             handler,
+		streamManager:       streamManager,
+	}
+}
+
+// GetHandlerForPath gets the handler for a path.
+func (m *PluginRunner) GetHandlerForPath(path string) (models.ChannelHandler, error) {
+	return &PluginPathRunner{
+		path:                path,
+		pluginID:            m.pluginID,
+		datasourceUID:       m.datasourceUID,
+		streamManager:       m.streamManager,
+		handler:             m.handler,
+		pluginContextGetter: m.pluginContextGetter,
+	}, nil
+}
+
+// PluginPathRunner can handle streaming operations for channels belonging to plugin specific path.
+type PluginPathRunner struct {
+	path                string
+	pluginID            string
+	datasourceUID       string
+	streamManager       *StreamManager
+	handler             backend.StreamHandler
+	pluginContextGetter PluginContextGetter
+}
+
+// OnSubscribe passes control to a plugin.
+func (r *PluginPathRunner) OnSubscribe(client *centrifuge.Client, e centrifuge.SubscribeEvent) (centrifuge.SubscribeReply, error) {
+	pCtx, found, err := r.pluginContextGetter.GetPluginContext(client.Context(), r.pluginID, r.datasourceUID)
+	if err != nil {
+		logger.Error("Get plugin context error", "error", err, "path", r.path)
+		return centrifuge.SubscribeReply{}, err
+	}
+	if !found {
+		logger.Error("Plugin context not found", "path", r.path)
+		return centrifuge.SubscribeReply{}, centrifuge.ErrorInternal
+	}
+	resp, err := r.handler.CanSubscribeToStream(client.Context(), &backend.SubscribeToStreamRequest{
+		PluginContext: pCtx,
+		Path:          r.path,
+	})
+	if err != nil {
+		logger.Error("Plugin CanSubscribeToStream call error", "error", err, "path", r.path)
+		return centrifuge.SubscribeReply{}, err
+	}
+	if !resp.OK {
+		return centrifuge.SubscribeReply{}, centrifuge.ErrorPermissionDenied
+	}
+	err = r.streamManager.SubmitStream(e.Channel, r.path, pCtx, r.handler)
+	if err != nil {
+		logger.Error("Error submitting stream to manager", "error", err, "path", r.path)
+		return centrifuge.SubscribeReply{}, centrifuge.ErrorInternal
+	}
+	return centrifuge.SubscribeReply{
+		Options: centrifuge.SubscribeOptions{
+			Presence: true,
+		},
+	}, nil
+}
+
+// OnPublish passes control to a plugin.
+func (r *PluginPathRunner) OnPublish(_ *centrifuge.Client, _ centrifuge.PublishEvent) (centrifuge.PublishReply, error) {
+	return centrifuge.PublishReply{}, fmt.Errorf("not implemented yet")
+}

pkg/services/live/features/stream.go

@@ -0,0 +1,173 @@
+package features
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"time"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+)
+
+// StreamManager manages streams from Grafana to plugins.
+type StreamManager struct {
+	mu               sync.RWMutex
+	streams          map[string]struct{}
+	presenceGetter   PresenceGetter
+	channelPublisher ChannelPublisher
+	registerCh       chan streamRequest
+	closedCh         chan struct{}
+	checkInterval    time.Duration
+	maxChecks        int
+}
+
+// StreamManagerOption modifies StreamManager behavior (used for tests for example).
+type StreamManagerOption func(*StreamManager)
+
+// WithCheckConfig allows setting custom check rules.
+func WithCheckConfig(interval time.Duration, maxChecks int) StreamManagerOption {
+	return func(sm *StreamManager) {
+		sm.checkInterval = interval
+		sm.maxChecks = maxChecks
+	}
+}
+
+const (
+	defaultCheckInterval = 5 * time.Second
+	defaultMaxChecks     = 3
+)
+
+// NewStreamManager creates new StreamManager.
+func NewStreamManager(chPublisher ChannelPublisher, presenceGetter PresenceGetter, opts ...StreamManagerOption) *StreamManager {
+	sm := &StreamManager{
+		streams:          make(map[string]struct{}),
+		channelPublisher: chPublisher,
+		presenceGetter:   presenceGetter,
+		registerCh:       make(chan streamRequest),
+		closedCh:         make(chan struct{}),
+		checkInterval:    defaultCheckInterval,
+		maxChecks:        defaultMaxChecks,
+	}
+	for _, opt := range opts {
+		opt(sm)
+	}
+	return sm
+}
+
+func (s *StreamManager) stopStream(sr streamRequest, cancelFn func()) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	delete(s.streams, sr.Channel)
+	cancelFn()
+}
+
+func (s *StreamManager) watchStream(ctx context.Context, cancelFn func(), sr streamRequest) {
+	numNoSubscribersChecks := 0
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(s.checkInterval):
+			numSubscribers, err := s.presenceGetter.GetNumSubscribers(sr.Channel)
+			if err != nil {
+				logger.Error("Error checking num subscribers", "channel", sr.Channel, "path", sr.Path)
+				continue
+			}
+			if numSubscribers > 0 {
+				// reset counter since channel has active subscribers.
+				numNoSubscribersChecks = 0
+				continue
+			}
+			numNoSubscribersChecks++
+			if numNoSubscribersChecks >= s.maxChecks {
+				logger.Info("Stop stream since no active subscribers", "channel", sr.Channel, "path", sr.Path)
+				s.stopStream(sr, cancelFn)
+				return
+			}
+		}
+	}
+}
+
+// run stream until context canceled.
+func (s *StreamManager) runStream(ctx context.Context, sr streamRequest) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+		err := sr.StreamRunner.RunStream(
+			ctx,
+			&backend.RunStreamRequest{
+				PluginContext: sr.PluginContext,
+				Path:          sr.Path,
+			},
+			newStreamSender(sr.Channel, s.channelPublisher),
+		)
+		if err != nil {
+			if errors.Is(ctx.Err(), context.Canceled) {
+				logger.Info("Stream cleanly finished", "path", sr.Path)
+				return
+			}
+			logger.Error("Error running stream, retrying", "path", sr.Path, "error", err)
+			continue
+		}
+		logger.Warn("Stream finished without error?", "path", sr.Path)
+		return
+	}
+}
+
+func (s *StreamManager) registerStream(ctx context.Context, sr streamRequest) {
+	s.mu.Lock()
+	if _, ok := s.streams[sr.Channel]; ok {
+		logger.Debug("Skip running new stream (already exists)", "path", sr.Path)
+		s.mu.Unlock()
+		return
+	}
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	s.streams[sr.Channel] = struct{}{}
+	s.mu.Unlock()
+
+	go s.watchStream(ctx, cancel, sr)
+	s.runStream(ctx, sr)
+}
+
+// Run StreamManager till context canceled.
+func (s *StreamManager) Run(ctx context.Context) error {
+	for {
+		select {
+		case sr := <-s.registerCh:
+			go s.registerStream(ctx, sr)
+		case <-ctx.Done():
+			close(s.closedCh)
+			return ctx.Err()
+		}
+	}
+}
+
+type streamRequest struct {
+	Channel       string
+	Path          string
+	PluginContext backend.PluginContext
+	StreamRunner  StreamRunner
+}
+
+// SubmitStream submits stream handler in StreamManager to manage.
+// The stream will be opened and kept till channel has active subscribers.
+func (s *StreamManager) SubmitStream(channel string, path string, pCtx backend.PluginContext, streamRunner StreamRunner) error {
+	select {
+	case <-s.closedCh:
+		close(s.registerCh)
+		return nil
+	case s.registerCh <- streamRequest{
+		Channel:       channel,
+		Path:          path,
+		PluginContext: pCtx,
+		StreamRunner:  streamRunner,
+	}:
+	case <-time.After(time.Second):
+		return errors.New("timeout")
+	}
+	return nil
+}

pkg/services/live/features/stream_test.go

@@ -0,0 +1,129 @@
+package features
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/require"
+)
+
+// wait until channel closed with timeout.
+func waitWithTimeout(tb testing.TB, ch chan struct{}, timeout time.Duration) {
+	select {
+	case <-ch:
+	case <-time.After(timeout):
+		tb.Fatal("timeout")
+	}
+}
+
+func TestStreamManager_Run(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+
+	mockChannelPublisher := NewMockChannelPublisher(mockCtrl)
+	mockPresenceGetter := NewMockPresenceGetter(mockCtrl)
+
+	manager := NewStreamManager(mockChannelPublisher, mockPresenceGetter)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	go func() {
+		cancel()
+	}()
+
+	err := manager.Run(ctx)
+	require.ErrorIs(t, err, context.Canceled)
+}
+
+func TestStreamManager_SubmitStream_Send(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+
+	mockChannelPublisher := NewMockChannelPublisher(mockCtrl)
+	mockPresenceGetter := NewMockPresenceGetter(mockCtrl)
+
+	manager := NewStreamManager(mockChannelPublisher, mockPresenceGetter)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go func() {
+		_ = manager.Run(ctx)
+	}()
+
+	startedCh := make(chan struct{})
+	doneCh := make(chan struct{})
+
+	mockChannelPublisher.EXPECT().Publish("test", []byte("test")).Times(1)
+
+	mockStreamRunner := NewMockStreamRunner(mockCtrl)
+	mockStreamRunner.EXPECT().RunStream(
+		gomock.Any(), gomock.Any(), gomock.Any(),
+	).Do(func(ctx context.Context, req *backend.RunStreamRequest, sender backend.StreamPacketSender) error {
+		require.Equal(t, "test", req.Path)
+		close(startedCh)
+		err := sender.Send(&backend.StreamPacket{
+			Payload: []byte("test"),
+		})
+		require.NoError(t, err)
+		<-ctx.Done()
+		close(doneCh)
+		return ctx.Err()
+	}).Times(1)
+
+	err := manager.SubmitStream("test", "test", backend.PluginContext{}, mockStreamRunner)
+	require.NoError(t, err)
+
+	// try submit the same.
+	err = manager.SubmitStream("test", "test", backend.PluginContext{}, mockStreamRunner)
+	require.NoError(t, err)
+
+	waitWithTimeout(t, startedCh, time.Second)
+	require.Len(t, manager.streams, 1)
+	cancel()
+	waitWithTimeout(t, doneCh, time.Second)
+}
+
+func TestStreamManager_SubmitStream_CloseNoSubscribers(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+
+	mockChannelPublisher := NewMockChannelPublisher(mockCtrl)
+	mockPresenceGetter := NewMockPresenceGetter(mockCtrl)
+
+	manager := NewStreamManager(
+		mockChannelPublisher,
+		mockPresenceGetter,
+		WithCheckConfig(10*time.Millisecond, 3),
+	)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go func() {
+		_ = manager.Run(ctx)
+	}()
+
+	startedCh := make(chan struct{})
+	doneCh := make(chan struct{})
+
+	mockPresenceGetter.EXPECT().GetNumSubscribers("test").Return(0, nil).Times(3)
+
+	mockStreamRunner := NewMockStreamRunner(mockCtrl)
+	mockStreamRunner.EXPECT().RunStream(gomock.Any(), gomock.Any(), gomock.Any()).Do(func(ctx context.Context, req *backend.RunStreamRequest, sender backend.StreamPacketSender) error {
+		close(startedCh)
+		<-ctx.Done()
+		close(doneCh)
+		return ctx.Err()
+	}).Times(1)
+
+	err := manager.SubmitStream("test", "test", backend.PluginContext{}, mockStreamRunner)
+	require.NoError(t, err)
+
+	waitWithTimeout(t, startedCh, time.Second)
+	waitWithTimeout(t, doneCh, time.Second)
+	require.Len(t, manager.streams, 0)
+}

pkg/services/live/features/testdata.go

@@ -1,113 +0,0 @@
-package features
-
-import (
-	"encoding/json"
-	"fmt"
-	"math/rand"
-	"time"
-
-	"github.com/centrifugal/centrifuge"
-	"github.com/grafana/grafana/pkg/models"
-)
-
-// testDataRunner manages all the `grafana/dashboard/*` channels.
-type testDataRunner struct {
-	publisher   models.ChannelPublisher
-	running     bool
-	speedMillis int
-	dropPercent float64
-	channel     string
-	name        string
-}
-
-// TestDataSupplier manages all the `grafana/testdata/*` channels.
-type TestDataSupplier struct {
-	Publisher models.ChannelPublisher
-}
-
-// GetHandlerForPath gets the channel handler for a path.
-// Called on init.
-func (s *TestDataSupplier) GetHandlerForPath(path string) (models.ChannelHandler, error) {
-	channel := "grafana/testdata/" + path
-
-	if path == "random-2s-stream" {
-		return &testDataRunner{
-			publisher:   s.Publisher,
-			running:     false,
-			speedMillis: 2000,
-			dropPercent: 0,
-			channel:     channel,
-			name:        path,
-		}, nil
-	}
-
-	if path == "random-flakey-stream" {
-		return &testDataRunner{
-			publisher:   s.Publisher,
-			running:     false,
-			speedMillis: 400,
-			dropPercent: .6,
-			channel:     channel,
-		}, nil
-	}
-
-	return nil, fmt.Errorf("unknown channel")
-}
-
-// OnSubscribe will let anyone connect to the path
-func (r *testDataRunner) OnSubscribe(c *centrifuge.Client, e centrifuge.SubscribeEvent) (centrifuge.SubscribeReply, error) {
-	if !r.running {
-		r.running = true
-
-		// Run in the background
-		go r.runRandomCSV()
-	}
-
-	return centrifuge.SubscribeReply{}, nil
-}
-
-// OnPublish checks if a message from the websocket can be broadcast on this channel
-func (r *testDataRunner) OnPublish(c *centrifuge.Client, e centrifuge.PublishEvent) (centrifuge.PublishReply, error) {
-	return centrifuge.PublishReply{}, fmt.Errorf("can not publish to testdata")
-}
-
-// runRandomCSV is just for an example.
-func (r *testDataRunner) runRandomCSV() {
-	spread := 50.0
-
-	walker := rand.Float64() * 100
-	ticker := time.NewTicker(time.Duration(r.speedMillis) * time.Millisecond)
-
-	measurement := models.Measurement{
-		Name:   r.name,
-		Time:   0,
-		Values: make(map[string]interface{}, 5),
-	}
-	msg := models.MeasurementBatch{
-		Measurements: []models.Measurement{measurement}, // always a single measurement
-	}
-
-	for t := range ticker.C {
-		if rand.Float64() <= r.dropPercent {
-			continue
-		}
-		delta := rand.Float64() - 0.5
-		walker += delta
-
-		measurement.Time = t.UnixNano() / int64(time.Millisecond)
-		measurement.Values["value"] = walker
-		measurement.Values["min"] = walker - ((rand.Float64() * spread) + 0.01)
-		measurement.Values["max"] = walker + ((rand.Float64() * spread) + 0.01)
-
-		bytes, err := json.Marshal(&msg)
-		if err != nil {
-			logger.Warn("unable to marshal line", "error", err)
-			continue
-		}
-
-		err = r.publisher(r.channel, bytes)
-		if err != nil {
-			logger.Warn("write", "channel", r.channel, "measurement", measurement)
-		}
-	}
-}

pkg/services/live/live.go

@@ -1,15 +1,21 @@
 package live
 
 import (
+	"context"
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/centrifugal/centrifuge"
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+
 	"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"
+	"github.com/grafana/grafana/pkg/plugins/manager"
+	"github.com/grafana/grafana/pkg/plugins/plugincontext"
 	"github.com/grafana/grafana/pkg/registry"
+	"github.com/grafana/grafana/pkg/services/datasources"
 	"github.com/grafana/grafana/pkg/services/live/features"
 	"github.com/grafana/grafana/pkg/setting"
 	"github.com/grafana/grafana/pkg/tsdb/cloudwatch"
@@ -21,13 +27,13 @@ var (
 )
 
 func init() {
-	registry.RegisterService(&GrafanaLive{
+	registry.RegisterServiceWithPriority(&GrafanaLive{
 		channels:   make(map[string]models.ChannelHandler),
 		channelsMu: sync.RWMutex{},
 		GrafanaScope: CoreGrafanaScope{
 			Features: make(map[string]models.ChannelHandlerFactory),
 		},
-	})
+	}, registry.Low)
 }
 
 // CoreGrafanaScope list of core features
@@ -40,11 +46,14 @@ type CoreGrafanaScope struct {
 
 // GrafanaLive pretends to be the server
 type GrafanaLive struct {
-	Cfg           *setting.Cfg            `inject:""`
-	RouteRegister routing.RouteRegister   `inject:""`
-	LogsService   *cloudwatch.LogsService `inject:""`
-	PluginManager plugins.Manager         `inject:""`
-	node          *centrifuge.Node
+	PluginContextProvider *plugincontext.Provider  `inject:""`
+	Cfg                   *setting.Cfg             `inject:""`
+	RouteRegister         routing.RouteRegister    `inject:""`
+	LogsService           *cloudwatch.LogsService  `inject:""`
+	PluginManager         *manager.PluginManager   `inject:""`
+	DatasourceCache       datasources.CacheService `inject:""`
+
+	node *centrifuge.Node
 
 	// The websocket handler
 	WebsocketHandler interface{}
@@ -55,9 +64,32 @@ type GrafanaLive struct {
 
 	// The core internal features
 	GrafanaScope CoreGrafanaScope
+
+	contextGetter *pluginContextGetter
+	streamManager *features.StreamManager
 }
 
-// Init initializes the instance.
+func (g *GrafanaLive) getStreamPlugin(pluginID string) (backend.StreamHandler, error) {
+	plugin, ok := g.PluginManager.BackendPluginManager.Get(pluginID)
+	if !ok {
+		return nil, fmt.Errorf("plugin not found: %s", pluginID)
+	}
+	streamHandler, ok := plugin.(backend.StreamHandler)
+	if !ok {
+		return nil, fmt.Errorf("%s plugin does not implement StreamHandler: %#v", pluginID, plugin)
+	}
+	return streamHandler, nil
+}
+
+func (g *GrafanaLive) Run(ctx context.Context) error {
+	if g.streamManager != nil {
+		// Only run stream manager if GrafanaLive properly initialized.
+		return g.streamManager.Run(ctx)
+	}
+	return nil
+}
+
+// Init initializes Live service.
 // Required to implement the registry.Service interface.
 func (g *GrafanaLive) Init() error {
 	logger.Debug("GrafanaLive initialization")
@@ -76,24 +108,25 @@ func (g *GrafanaLive) Init() error {
 
 	// Node is the core object in Centrifuge library responsible for many useful
 	// things. For example Node allows to publish messages to channels from server
-	// side with its Publish method, but in this example we will publish messages
-	// only from client side.
+	// side with its Publish method.
 	node, err := centrifuge.New(cfg)
 	if err != nil {
 		return err
 	}
 	g.node = node
 
+	g.contextGetter = newPluginContextGetter(g.PluginContextProvider)
+
+	channelPublisher := newPluginChannelPublisher(node)
+	presenceGetter := newPluginPresenceGetter(node)
+	g.streamManager = features.NewStreamManager(channelPublisher, presenceGetter)
+
 	// Initialize the main features
 	dash := &features.DashboardHandler{
 		Publisher: g.Publish,
 	}
-
 	g.GrafanaScope.Dashboards = dash
 	g.GrafanaScope.Features["dashboard"] = dash
-	g.GrafanaScope.Features["testdata"] = &features.TestDataSupplier{
-		Publisher: g.Publish,
-	}
 	g.GrafanaScope.Features["broadcast"] = &features.BroadcastRunner{}
 	g.GrafanaScope.Features["measurements"] = &features.MeasurementsRunner{}
 
@@ -102,11 +135,14 @@ func (g *GrafanaLive) Init() error {
 	// different goroutines (belonging to different client connections). This is also
 	// true for other event handlers.
 	node.OnConnect(func(client *centrifuge.Client) {
-		logger.Debug("Client connected", "user", client.UserID())
+		logger.Debug("Client connected", "user", client.UserID(), "client", client.ID())
+		connectedAt := time.Now()
 
 		client.OnSubscribe(func(e centrifuge.SubscribeEvent, cb centrifuge.SubscribeCallback) {
-			handler, err := g.GetChannelHandler(e.Channel)
+			logger.Debug("Client wants to subscribe", "user", client.UserID(), "client", client.ID(), "channel", e.Channel)
+			handler, err := g.GetChannelHandler(client.Context(), e.Channel)
 			if err != nil {
+				logger.Error("Error getting channel handler", "user", client.UserID(), "client", client.ID(), "channel", e.Channel, "error", err)
 				cb(centrifuge.SubscribeReply{}, err)
 			} else {
 				cb(handler.OnSubscribe(client, e))
@@ -117,13 +153,19 @@ func (g *GrafanaLive) Init() error {
 		// In general, we should prefer writing to the HTTP API, but this
 		// allows some simple prototypes to work quickly.
 		client.OnPublish(func(e centrifuge.PublishEvent, cb centrifuge.PublishCallback) {
-			handler, err := g.GetChannelHandler(e.Channel)
+			logger.Debug("Client wants to publish", "user", client.UserID(), "client", client.ID(), "channel", e.Channel)
+			handler, err := g.GetChannelHandler(client.Context(), e.Channel)
 			if err != nil {
+				logger.Error("Error getting channel handler", "user", client.UserID(), "client", client.ID(), "channel", e.Channel, "error", err)
 				cb(centrifuge.PublishReply{}, err)
 			} else {
 				cb(handler.OnPublish(client, e))
 			}
 		})
+
+		client.OnDisconnect(func(_ centrifuge.DisconnectEvent) {
+			logger.Debug("Client disconnected", "user", client.UserID(), "client", client.ID(), "elapsed", time.Since(connectedAt))
+		})
 	})
 
 	// Run node. This method does not block.
@@ -149,6 +191,7 @@ func (g *GrafanaLive) Init() error {
 			UserID: fmt.Sprintf("%d", user.UserId),
 		}
 		newCtx := centrifuge.SetCredentials(ctx.Req.Context(), cred)
+		newCtx = setContextSignedUser(newCtx, user)
 
 		r := ctx.Req.Request
 		r = r.WithContext(newCtx) // Set a user ID.
@@ -161,12 +204,13 @@ func (g *GrafanaLive) Init() error {
 	return nil
 }
 
-// GetChannelHandler gives threadsafe access to the channel
-func (g *GrafanaLive) GetChannelHandler(channel string) (models.ChannelHandler, error) {
+// GetChannelHandler gives thread-safe access to the channel.
+func (g *GrafanaLive) GetChannelHandler(ctx context.Context, channel string) (models.ChannelHandler, error) {
 	g.channelsMu.RLock()
 	c, ok := g.channels[channel]
 	g.channelsMu.RUnlock() // defer? but then you can't lock further down
 	if ok {
+		logger.Debug("Found cached channel handler", "channel", channel)
 		return c, nil
 	}
 
@@ -175,65 +219,94 @@ func (g *GrafanaLive) GetChannelHandler(channel string) (models.ChannelHandler,
 	if !addr.IsValid() {
 		return nil, fmt.Errorf("invalid channel: %q", channel)
 	}
-	logger.Info("initChannel", "channel", channel, "address", addr)
 
 	g.channelsMu.Lock()
 	defer g.channelsMu.Unlock()
 	c, ok = g.channels[channel] // may have filled in while locked
 	if ok {
+		logger.Debug("Found cached channel handler", "channel", channel)
 		return c, nil
 	}
 
-	getter, err := g.GetChannelHandlerFactory(addr.Scope, addr.Namespace)
+	getter, err := g.GetChannelHandlerFactory(ctx, addr.Scope, addr.Namespace)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("error getting channel handler factory: %w", err)
 	}
 
-	// First access will initialize
+	// First access will initialize.
 	c, err = getter.GetHandlerForPath(addr.Path)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("error getting handler for path: %w", err)
 	}
 
+	logger.Info("Initialized channel handler", "channel", channel, "address", addr)
 	g.channels[channel] = c
 	return c, nil
 }
 
 // GetChannelHandlerFactory gets a ChannelHandlerFactory for a namespace.
-// It gives threadsafe access to the channel.
-func (g *GrafanaLive) GetChannelHandlerFactory(scope string, name string) (models.ChannelHandlerFactory, error) {
-	if scope == "grafana" {
-		p, ok := g.GrafanaScope.Features[name]
-		if ok {
-			return p, nil
-		}
-		return nil, fmt.Errorf("unknown feature: %q", name)
+// It gives thread-safe access to the channel.
+func (g *GrafanaLive) GetChannelHandlerFactory(ctx context.Context, scope string, namespace string) (models.ChannelHandlerFactory, error) {
+	switch scope {
+	case ScopeGrafana:
+		return g.handleGrafanaScope(ctx, namespace)
+	case ScopePlugin:
+		return g.handlePluginScope(ctx, namespace)
+	case ScopeDatasource:
+		return g.handleDatasourceScope(ctx, namespace)
+	default:
+		return nil, fmt.Errorf("invalid scope: %q", scope)
 	}
+}
 
-	if scope == "ds" {
-		return nil, fmt.Errorf("todo... look up datasource: %q", name)
+func (g *GrafanaLive) handleGrafanaScope(_ context.Context, namespace string) (models.ChannelHandlerFactory, error) {
+	if p, ok := g.GrafanaScope.Features[namespace]; ok {
+		return p, nil
 	}
+	return nil, fmt.Errorf("unknown feature: %q", namespace)
+}
 
-	if scope == "plugin" {
-		// Temporary hack until we have a more generic solution later on
-		if name == "cloudwatch" {
-			return &cloudwatch.LogQueryRunnerSupplier{
-				Publisher: g.Publish,
-				Service:   g.LogsService,
-			}, nil
-		}
-
-		p := g.PluginManager.GetPlugin(name)
-		if p != nil {
-			h := &PluginHandler{
-				Plugin: p,
-			}
-			return h, nil
-		}
-		return nil, fmt.Errorf("unknown plugin: %q", name)
+func (g *GrafanaLive) handlePluginScope(_ context.Context, namespace string) (models.ChannelHandlerFactory, error) {
+	// Temporary hack until we have a more generic solution later on
+	if namespace == "cloudwatch" {
+		return &cloudwatch.LogQueryRunnerSupplier{
+			Publisher: g.Publish,
+			Service:   g.LogsService,
+		}, nil
 	}
+	streamHandler, err := g.getStreamPlugin(namespace)
+	if err != nil {
+		return nil, fmt.Errorf("can't find stream plugin: %s", namespace)
+	}
+	return features.NewPluginRunner(
+		namespace,
+		"",
+		g.streamManager,
+		g.contextGetter,
+		streamHandler,
+	), nil
+}
 
-	return nil, fmt.Errorf("invalid scope: %q", scope)
+func (g *GrafanaLive) handleDatasourceScope(ctx context.Context, namespace string) (models.ChannelHandlerFactory, error) {
+	user, ok := getContextSignedUser(ctx)
+	if !ok {
+		return nil, fmt.Errorf("no signed user found in context")
+	}
+	ds, err := g.DatasourceCache.GetDatasourceByUID(namespace, user, false)
+	if err != nil {
+		return nil, fmt.Errorf("error getting datasource: %w", err)
+	}
+	streamHandler, err := g.getStreamPlugin(ds.Name)
+	if err != nil {
+		return nil, fmt.Errorf("can't find stream plugin: %s", namespace)
+	}
+	return features.NewPluginRunner(
+		ds.Type,
+		ds.Uid,
+		g.streamManager,
+		g.contextGetter,
+		streamHandler,
+	), nil
 }
 
 // Publish sends the data to the channel without checking permissions etc

pkg/services/live/pluginHandler.go

@@ -1,27 +0,0 @@
-package live
-
-import (
-	"github.com/centrifugal/centrifuge"
-	"github.com/grafana/grafana/pkg/models"
-	"github.com/grafana/grafana/pkg/plugins"
-)
-
-// PluginHandler manages all the `grafana/dashboard/*` channels
-type PluginHandler struct {
-	Plugin *plugins.PluginBase
-}
-
-// GetHandlerForPath called on init
-func (h *PluginHandler) GetHandlerForPath(path string) (models.ChannelHandler, error) {
-	return h, nil // all dashboards share the same handler
-}
-
-// OnSubscribe for now allows anyone to subscribe
-func (h *PluginHandler) OnSubscribe(c *centrifuge.Client, e centrifuge.SubscribeEvent) (centrifuge.SubscribeReply, error) {
-	return centrifuge.SubscribeReply{}, nil
-}
-
-// OnPublish checks if a message from the websocket can be broadcast on this channel
-func (h *PluginHandler) OnPublish(c *centrifuge.Client, e centrifuge.PublishEvent) (centrifuge.PublishReply, error) {
-	return centrifuge.PublishReply{}, nil // broadcast any event
-}

pkg/services/live/plugin_helpers.go

@@ -0,0 +1,57 @@
+package live
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/centrifugal/centrifuge"
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"github.com/grafana/grafana/pkg/plugins/plugincontext"
+)
+
+type pluginChannelPublisher struct {
+	node *centrifuge.Node
+}
+
+func newPluginChannelPublisher(node *centrifuge.Node) *pluginChannelPublisher {
+	return &pluginChannelPublisher{node: node}
+}
+
+func (p *pluginChannelPublisher) Publish(channel string, data []byte) error {
+	_, err := p.node.Publish(channel, data)
+	return err
+}
+
+type pluginPresenceGetter struct {
+	node *centrifuge.Node
+}
+
+func newPluginPresenceGetter(node *centrifuge.Node) *pluginPresenceGetter {
+	return &pluginPresenceGetter{node: node}
+}
+
+func (p *pluginPresenceGetter) GetNumSubscribers(channel string) (int, error) {
+	res, err := p.node.PresenceStats(channel)
+	if err != nil {
+		return 0, err
+	}
+	return res.NumClients, nil
+}
+
+type pluginContextGetter struct {
+	PluginContextProvider *plugincontext.Provider
+}
+
+func newPluginContextGetter(pluginContextProvider *plugincontext.Provider) *pluginContextGetter {
+	return &pluginContextGetter{
+		PluginContextProvider: pluginContextProvider,
+	}
+}
+
+func (g *pluginContextGetter) GetPluginContext(ctx context.Context, pluginID string, datasourceUID string) (backend.PluginContext, bool, error) {
+	user, ok := getContextSignedUser(ctx)
+	if !ok {
+		return backend.PluginContext{}, false, fmt.Errorf("no signed user found in context")
+	}
+	return g.PluginContextProvider.Get(pluginID, datasourceUID, user)
+}

pkg/services/live/scope.go

@@ -0,0 +1,10 @@
+package live
+
+const (
+	// ScopeGrafana contains builtin features of Grafana Core.
+	ScopeGrafana = "grafana"
+	// ScopePlugin passes control to a plugin.
+	ScopePlugin = "plugin"
+	// ScopeDatasource passes control to a datasource plugin.
+	ScopeDatasource = "ds"
+)