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

Live: simplify dependencies (#115130)

Browse Source
This commit is contained in:
Ryan McKinley
2025-12-11 13:37:45 +03:00
committed by GitHub
parent 7fe3214f16
commit 9bec62a080
26 changed files with 93 additions and 522 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/services/live/database/storage.go
-48
View File
@@ -1,48 +0,0 @@
package database
import (
"fmt"
"time"
"github.com/grafana/grafana/pkg/infra/db"
"github.com/grafana/grafana/pkg/infra/localcache"
"github.com/grafana/grafana/pkg/services/live/model"
)
type Storage struct {
store db.DB
cache *localcache.CacheService
}
func NewStorage(store db.DB, cache *localcache.CacheService) *Storage {
return &Storage{store: store, cache: cache}
}
func getLiveMessageCacheKey(orgID int64, channel string) string {
return fmt.Sprintf("live_message_%d_%s", orgID, channel)
}
func (s *Storage) SaveLiveMessage(query *model.SaveLiveMessageQuery) error {
// Come back to saving into database after evaluating database structure.
s.cache.Set(getLiveMessageCacheKey(query.OrgID, query.Channel), model.LiveMessage{
ID: 0, // Not used actually.
OrgID: query.OrgID,
Channel: query.Channel,
Data: query.Data,
Published: time.Now(),
}, 0)
return nil
}
func (s *Storage) GetLiveMessage(query *model.GetLiveMessageQuery) (model.LiveMessage, bool, error) {
// Come back to saving into database after evaluating database structure.
m, ok := s.cache.Get(getLiveMessageCacheKey(query.OrgID, query.Channel))
if !ok {
return model.LiveMessage{}, false, nil
}
msg, ok := m.(model.LiveMessage)
if !ok {
return model.LiveMessage{}, false, fmt.Errorf("unexpected live message type in cache: %T", m)
}
return msg, true, nil
}
pkg/services/live/database/tests/setup.go
-18
View File
@@ -1,18 +0,0 @@
package tests
import (
"testing"
"time"
"github.com/grafana/grafana/pkg/infra/db"
"github.com/grafana/grafana/pkg/infra/localcache"
"github.com/grafana/grafana/pkg/services/live/database"
)
// SetupTestStorage initializes a storage to used by the integration tests.
// This is required to properly register and execute migrations.
func SetupTestStorage(t *testing.T) *database.Storage {
sqlStore := db.InitTestDB(t)
localCache := localcache.New(time.Hour, time.Hour)
return database.NewStorage(sqlStore, localCache)
}
pkg/services/live/database/tests/storage_test.go
-67
View File
@@ -1,67 +0,0 @@
package tests
import (
"encoding/json"
"testing"
"github.com/stretchr/testify/require"
"github.com/grafana/grafana/pkg/services/live/model"
"github.com/grafana/grafana/pkg/tests/testsuite"
"github.com/grafana/grafana/pkg/util/testutil"
)
func TestMain(m *testing.M) {
testsuite.Run(m)
}
func TestIntegrationLiveMessage(t *testing.T) {
testutil.SkipIntegrationTestInShortMode(t)
storage := SetupTestStorage(t)
getQuery := &model.GetLiveMessageQuery{
OrgID: 1,
Channel: "test_channel",
}
_, ok, err := storage.GetLiveMessage(getQuery)
require.NoError(t, err)
require.False(t, ok)
saveQuery := &model.SaveLiveMessageQuery{
OrgID: 1,
Channel: "test_channel",
Data: []byte(`{}`),
}
err = storage.SaveLiveMessage(saveQuery)
require.NoError(t, err)
msg, ok, err := storage.GetLiveMessage(getQuery)
require.NoError(t, err)
require.True(t, ok)
require.Equal(t, int64(1), msg.OrgID)
require.Equal(t, "test_channel", msg.Channel)
require.Equal(t, json.RawMessage(`{}`), msg.Data)
require.NotZero(t, msg.Published)
// try saving again, should be replaced.
saveQuery2 := &model.SaveLiveMessageQuery{
OrgID: 1,
Channel: "test_channel",
Data: []byte(`{"input": "hello"}`),
}
err = storage.SaveLiveMessage(saveQuery2)
require.NoError(t, err)
getQuery2 := &model.GetLiveMessageQuery{
OrgID: 1,
Channel: "test_channel",
}
msg2, ok, err := storage.GetLiveMessage(getQuery2)
require.NoError(t, err)
require.True(t, ok)
require.Equal(t, int64(1), msg2.OrgID)
require.Equal(t, "test_channel", msg2.Channel)
require.Equal(t, json.RawMessage(`{"input": "hello"}`), msg2.Data)
require.NotZero(t, msg2.Published)
}
pkg/services/live/features/broadcast.go
-70
View File
@@ -1,70 +0,0 @@
package features
import (
"context"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana/pkg/apimachinery/identity"
"github.com/grafana/grafana/pkg/infra/log"
"github.com/grafana/grafana/pkg/services/live/model"
)
var (
logger = log.New("live.features") // scoped to all features?
)
//go:generate mockgen -destination=broadcast_mock.go -package=features github.com/grafana/grafana/pkg/services/live/features LiveMessageStore
type LiveMessageStore interface {
SaveLiveMessage(query *model.SaveLiveMessageQuery) error
GetLiveMessage(query *model.GetLiveMessageQuery) (model.LiveMessage, bool, error)
}
// BroadcastRunner will simply broadcast all events to `grafana/broadcast/*` channels
// This assumes that data is a JSON object
type BroadcastRunner struct {
liveMessageStore LiveMessageStore
}
func NewBroadcastRunner(liveMessageStore LiveMessageStore) *BroadcastRunner {
return &BroadcastRunner{liveMessageStore: liveMessageStore}
}
// GetHandlerForPath called on init
func (b *BroadcastRunner) GetHandlerForPath(_ string) (model.ChannelHandler, error) {
return b, nil // all dashboards share the same handler
}
// OnSubscribe will let anyone connect to the path
func (b *BroadcastRunner) OnSubscribe(_ context.Context, u identity.Requester, e model.SubscribeEvent) (model.SubscribeReply, backend.SubscribeStreamStatus, error) {
reply := model.SubscribeReply{
Presence: true,
JoinLeave: true,
}
query := &model.GetLiveMessageQuery{
OrgID: u.GetOrgID(),
Channel: e.Channel,
}
msg, ok, err := b.liveMessageStore.GetLiveMessage(query)
if err != nil {
return model.SubscribeReply{}, 0, err
}
if ok {
reply.Data = msg.Data
}
return reply, backend.SubscribeStreamStatusOK, nil
}
// OnPublish is called when a client wants to broadcast on the websocket
func (b *BroadcastRunner) OnPublish(_ context.Context, u identity.Requester, e model.PublishEvent) (model.PublishReply, backend.PublishStreamStatus, error) {
query := &model.SaveLiveMessageQuery{
OrgID: u.GetOrgID(),
Channel: e.Channel,
Data: e.Data,
}
if err := b.liveMessageStore.SaveLiveMessage(query); err != nil {
return model.PublishReply{}, 0, err
}
return model.PublishReply{Data: e.Data}, backend.PublishStreamStatusOK, nil
}
pkg/services/live/features/broadcast_mock.go
-66
View File
@@ -1,66 +0,0 @@
// Code generated by MockGen. DO NOT EDIT.
// Source: github.com/grafana/grafana/pkg/services/live/features (interfaces: LiveMessageStore)
// Package features is a generated GoMock package.
package features
import (
reflect "reflect"
gomock "github.com/golang/mock/gomock"
model "github.com/grafana/grafana/pkg/services/live/model"
)
// MockLiveMessageStore is a mock of LiveMessageStore interface.
type MockLiveMessageStore struct {
ctrl *gomock.Controller
recorder *MockLiveMessageStoreMockRecorder
}
// MockLiveMessageStoreMockRecorder is the mock recorder for MockLiveMessageStore.
type MockLiveMessageStoreMockRecorder struct {
mock *MockLiveMessageStore
}
// NewMockLiveMessageStore creates a new mock instance.
func NewMockLiveMessageStore(ctrl *gomock.Controller) *MockLiveMessageStore {
mock := &MockLiveMessageStore{ctrl: ctrl}
mock.recorder = &MockLiveMessageStoreMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
func (m *MockLiveMessageStore) EXPECT() *MockLiveMessageStoreMockRecorder {
return m.recorder
}
// GetLiveMessage mocks base method.
func (m *MockLiveMessageStore) GetLiveMessage(arg0 *model.GetLiveMessageQuery) (model.LiveMessage, bool, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetLiveMessage", arg0)
ret0, _ := ret[0].(model.LiveMessage)
ret1, _ := ret[1].(bool)
ret2, _ := ret[2].(error)
return ret0, ret1, ret2
}
// GetLiveMessage indicates an expected call of GetLiveMessage.
func (mr *MockLiveMessageStoreMockRecorder) GetLiveMessage(arg0 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetLiveMessage", reflect.TypeOf((*MockLiveMessageStore)(nil).GetLiveMessage), arg0)
}
// SaveLiveMessage mocks base method.
func (m *MockLiveMessageStore) SaveLiveMessage(arg0 *model.SaveLiveMessageQuery) error {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "SaveLiveMessage", arg0)
ret0, _ := ret[0].(error)
return ret0
}
// SaveLiveMessage indicates an expected call of SaveLiveMessage.
func (mr *MockLiveMessageStoreMockRecorder) SaveLiveMessage(arg0 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SaveLiveMessage", reflect.TypeOf((*MockLiveMessageStore)(nil).SaveLiveMessage), arg0)
}
pkg/services/live/features/broadcast_test.go
-87
View File
@@ -1,87 +0,0 @@
package features
import (
"context"
"encoding/json"
"testing"
"github.com/golang/mock/gomock"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/stretchr/testify/require"
"github.com/grafana/grafana/pkg/services/live/model"
"github.com/grafana/grafana/pkg/services/user"
)
func TestNewBroadcastRunner(t *testing.T) {
mockCtrl := gomock.NewController(t)
defer mockCtrl.Finish()
d := NewMockLiveMessageStore(mockCtrl)
br := NewBroadcastRunner(d)
require.NotNil(t, br)
}
func TestBroadcastRunner_OnSubscribe(t *testing.T) {
mockCtrl := gomock.NewController(t)
defer mockCtrl.Finish()
mockDispatcher := NewMockLiveMessageStore(mockCtrl)
channel := "stream/channel/test"
data := json.RawMessage(`{}`)
mockDispatcher.EXPECT().GetLiveMessage(&model.GetLiveMessageQuery{
OrgID: 1,
Channel: channel,
}).DoAndReturn(func(query *model.GetLiveMessageQuery) (model.LiveMessage, bool, error) {
return model.LiveMessage{
Data: data,
}, true, nil
}).Times(1)
br := NewBroadcastRunner(mockDispatcher)
require.NotNil(t, br)
handler, err := br.GetHandlerForPath("test")
require.NoError(t, err)
reply, status, err := handler.OnSubscribe(
context.Background(),
&user.SignedInUser{OrgID: 1, UserID: 2},
model.SubscribeEvent{Channel: channel, Path: "test"},
)
require.NoError(t, err)
require.Equal(t, backend.SubscribeStreamStatusOK, status)
require.Equal(t, data, reply.Data)
require.True(t, reply.Presence)
require.True(t, reply.JoinLeave)
require.False(t, reply.Recover)
}
func TestBroadcastRunner_OnPublish(t *testing.T) {
mockCtrl := gomock.NewController(t)
defer mockCtrl.Finish()
mockDispatcher := NewMockLiveMessageStore(mockCtrl)
channel := "stream/channel/test"
data := json.RawMessage(`{}`)
var orgID int64 = 1
mockDispatcher.EXPECT().SaveLiveMessage(&model.SaveLiveMessageQuery{
OrgID: orgID,
Channel: channel,
Data: data,
}).DoAndReturn(func(query *model.SaveLiveMessageQuery) error {
return nil
}).Times(1)
br := NewBroadcastRunner(mockDispatcher)
require.NotNil(t, br)
handler, err := br.GetHandlerForPath("test")
require.NoError(t, err)
reply, status, err := handler.OnPublish(
context.Background(),
&user.SignedInUser{OrgID: 1, UserID: 2},
model.PublishEvent{Channel: channel, Path: "test", Data: data},
)
require.NoError(t, err)
require.Equal(t, backend.PublishStreamStatusOK, status)
require.Equal(t, data, reply.Data)
}
pkg/services/live/features/dashboard.go
+1 -3
View File
@@ -7,9 +7,8 @@ import (
"strings"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana/pkg/apimachinery/identity"
"github.com/grafana/grafana/pkg/infra/db"
"github.com/grafana/grafana/pkg/cmd/grafana-cli/logger"
"github.com/grafana/grafana/pkg/services/accesscontrol"
"github.com/grafana/grafana/pkg/services/dashboards"
"github.com/grafana/grafana/pkg/services/live/model"
@@ -35,7 +34,6 @@ type dashboardEvent struct {
type DashboardHandler struct {
Publisher model.ChannelPublisher
ClientCount model.ChannelClientCount
Store db.DB
DashboardService dashboards.DashboardService
AccessControl accesscontrol.AccessControl
}
pkg/services/live/features/plugin.go
+2
View File
@@ -5,9 +5,11 @@ import (
"errors"
"github.com/centrifugal/centrifuge"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana/pkg/apimachinery/identity"
"github.com/grafana/grafana/pkg/cmd/grafana-cli/logger"
"github.com/grafana/grafana/pkg/plugins"
"github.com/grafana/grafana/pkg/services/live/model"
"github.com/grafana/grafana/pkg/services/live/orgchannel"
pkg/services/live/live.go
+17 -59
View File
@@ -15,7 +15,6 @@ import (
"github.com/centrifugal/centrifuge"
"github.com/gobwas/glob"
jsoniter "github.com/json-iterator/go"
"github.com/redis/go-redis/v9"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
@@ -25,12 +24,9 @@ import (
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana-plugin-sdk-go/live"
"github.com/grafana/grafana/pkg/api/dtos"
"github.com/grafana/grafana/pkg/api/response"
"github.com/grafana/grafana/pkg/api/routing"
"github.com/grafana/grafana/pkg/apimachinery/errutil"
"github.com/grafana/grafana/pkg/apimachinery/identity"
"github.com/grafana/grafana/pkg/infra/db"
"github.com/grafana/grafana/pkg/infra/localcache"
"github.com/grafana/grafana/pkg/infra/log"
"github.com/grafana/grafana/pkg/infra/usagestats"
@@ -43,7 +39,6 @@ import (
"github.com/grafana/grafana/pkg/services/dashboards"
"github.com/grafana/grafana/pkg/services/datasources"
"github.com/grafana/grafana/pkg/services/featuremgmt"
"github.com/grafana/grafana/pkg/services/live/database"
"github.com/grafana/grafana/pkg/services/live/features"
"github.com/grafana/grafana/pkg/services/live/livecontext"
"github.com/grafana/grafana/pkg/services/live/liveplugin"
@@ -57,7 +52,6 @@ import (
"github.com/grafana/grafana/pkg/services/org"
"github.com/grafana/grafana/pkg/services/pluginsintegration/plugincontext"
"github.com/grafana/grafana/pkg/services/pluginsintegration/pluginstore"
"github.com/grafana/grafana/pkg/services/query"
"github.com/grafana/grafana/pkg/services/secrets"
"github.com/grafana/grafana/pkg/setting"
"github.com/grafana/grafana/pkg/util"
@@ -80,8 +74,8 @@ type CoreGrafanaScope struct {
func ProvideService(plugCtxProvider *plugincontext.Provider, cfg *setting.Cfg, routeRegister routing.RouteRegister,
pluginStore pluginstore.Store, pluginClient plugins.Client, cacheService *localcache.CacheService,
dataSourceCache datasources.CacheService, sqlStore db.DB, secretsService secrets.Service,
usageStatsService usagestats.Service, queryDataService query.Service, toggles featuremgmt.FeatureToggles,
dataSourceCache datasources.CacheService, secretsService secrets.Service,
usageStatsService usagestats.Service, toggles featuremgmt.FeatureToggles,
accessControl accesscontrol.AccessControl, dashboardService dashboards.DashboardService,
orgService org.Service, configProvider apiserver.RestConfigProvider) (*GrafanaLive, error) {
g := &GrafanaLive{
@@ -93,9 +87,7 @@ func ProvideService(plugCtxProvider *plugincontext.Provider, cfg *setting.Cfg, r
pluginClient: pluginClient,
CacheService: cacheService,
DataSourceCache: dataSourceCache,
SQLStore: sqlStore,
SecretsService: secretsService,
queryDataService: queryDataService,
channels: make(map[string]model.ChannelHandler),
GrafanaScope: CoreGrafanaScope{
Features: make(map[string]model.ChannelHandlerFactory),
@@ -186,14 +178,11 @@ func ProvideService(plugCtxProvider *plugincontext.Provider, cfg *setting.Cfg, r
dash := &features.DashboardHandler{
Publisher: g.Publish,
ClientCount: g.ClientCount,
Store: sqlStore,
DashboardService: dashboardService,
AccessControl: accessControl,
}
g.storage = database.NewStorage(g.SQLStore, g.CacheService)
g.GrafanaScope.Dashboards = dash
g.GrafanaScope.Features["dashboard"] = dash
g.GrafanaScope.Features["broadcast"] = features.NewBroadcastRunner(g.storage)
// Testing watch with just the provisioning support -- this will be removed when it is well validated
//nolint:staticcheck // not yet migrated to OpenFeature
@@ -388,14 +377,14 @@ func ProvideService(plugCtxProvider *plugincontext.Provider, cfg *setting.Cfg, r
UserID: strconv.FormatInt(id, 10),
}
newCtx := centrifuge.SetCredentials(ctx.Req.Context(), cred)
newCtx = livecontext.SetContextSignedUser(newCtx, user)
newCtx = identity.WithRequester(newCtx, user)
r := ctx.Req.WithContext(newCtx)
wsHandler.ServeHTTP(ctx.Resp, r)
}
g.pushWebsocketHandler = func(ctx *contextmodel.ReqContext) {
user := ctx.SignedInUser
newCtx := livecontext.SetContextSignedUser(ctx.Req.Context(), user)
newCtx := identity.WithRequester(ctx.Req.Context(), user)
newCtx = livecontext.SetContextStreamID(newCtx, web.Params(ctx.Req)[":streamId"])
r := ctx.Req.WithContext(newCtx)
pushWSHandler.ServeHTTP(ctx.Resp, r)
@@ -403,7 +392,7 @@ func ProvideService(plugCtxProvider *plugincontext.Provider, cfg *setting.Cfg, r
g.pushPipelineWebsocketHandler = func(ctx *contextmodel.ReqContext) {
user := ctx.SignedInUser
newCtx := livecontext.SetContextSignedUser(ctx.Req.Context(), user)
newCtx := identity.WithRequester(ctx.Req.Context(), user)
newCtx = livecontext.SetContextChannelID(newCtx, web.Params(ctx.Req)["*"])
r := ctx.Req.WithContext(newCtx)
pushPipelineWSHandler.ServeHTTP(ctx.Resp, r)
@@ -475,14 +464,12 @@ type GrafanaLive struct {
RouteRegister routing.RouteRegister
CacheService *localcache.CacheService
DataSourceCache datasources.CacheService
SQLStore db.DB
SecretsService secrets.Service
pluginStore pluginstore.Store
pluginClient plugins.Client
queryDataService query.Service
orgService org.Service
keyPrefix string
keyPrefix string // HA prefix for grafana cloud (since the org is always 1)
node *centrifuge.Node
surveyCaller *survey.Caller
@@ -505,7 +492,6 @@ type GrafanaLive struct {
contextGetter *liveplugin.ContextGetter
runStreamManager *runstream.Manager
storage *database.Storage
usageStatsService usagestats.Service
usageStats usageStats
@@ -673,18 +659,13 @@ func (g *GrafanaLive) HandleDatasourceUpdate(orgID int64, dsUID string) {
}
}
// Use a configuration that's compatible with the standard library
// to minimize the risk of introducing bugs. This will make sure
// that map keys is ordered.
var jsonStd = jsoniter.ConfigCompatibleWithStandardLibrary
func (g *GrafanaLive) handleOnRPC(clientContextWithSpan context.Context, client *centrifuge.Client, e centrifuge.RPCEvent) (centrifuge.RPCReply, error) {
logger.Debug("Client calls RPC", "user", client.UserID(), "client", client.ID(), "method", e.Method)
if e.Method != "grafana.query" {
return centrifuge.RPCReply{}, centrifuge.ErrorMethodNotFound
}
user, ok := livecontext.GetContextSignedUser(clientContextWithSpan)
if !ok {
user, err := identity.GetRequester(clientContextWithSpan)
if err != nil {
logger.Error("No user found in context", "user", client.UserID(), "client", client.ID(), "method", e.Method)
return centrifuge.RPCReply{}, centrifuge.ErrorInternal
}
@@ -694,38 +675,15 @@ func (g *GrafanaLive) handleOnRPC(clientContextWithSpan context.Context, client
return centrifuge.RPCReply{}, centrifuge.ErrorExpired
}
var req dtos.MetricRequest
err := json.Unmarshal(e.Data, &req)
if err != nil {
return centrifuge.RPCReply{}, centrifuge.ErrorBadRequest
}
resp, err := g.queryDataService.QueryData(clientContextWithSpan, user, false, req)
if err != nil {
logger.Error("Error query data", "user", client.UserID(), "client", client.ID(), "method", e.Method, "error", err)
if errors.Is(err, datasources.ErrDataSourceAccessDenied) {
return centrifuge.RPCReply{}, &centrifuge.Error{Code: uint32(http.StatusForbidden), Message: http.StatusText(http.StatusForbidden)}
}
var gfErr errutil.Error
if errors.As(err, &gfErr) && gfErr.Reason.Status() == errutil.StatusBadRequest {
return centrifuge.RPCReply{}, &centrifuge.Error{Code: uint32(http.StatusBadRequest), Message: http.StatusText(http.StatusBadRequest)}
}
return centrifuge.RPCReply{}, centrifuge.ErrorInternal
}
data, err := jsonStd.Marshal(resp)
if err != nil {
logger.Error("Error marshaling query response", "user", client.UserID(), "client", client.ID(), "method", e.Method, "error", err)
return centrifuge.RPCReply{}, centrifuge.ErrorInternal
}
return centrifuge.RPCReply{
Data: data,
}, nil
// RPC events not available
return centrifuge.RPCReply{}, centrifuge.ErrorNotAvailable
}
func (g *GrafanaLive) handleOnSubscribe(clientContextWithSpan context.Context, client *centrifuge.Client, e centrifuge.SubscribeEvent) (centrifuge.SubscribeReply, error) {
logger.Debug("Client wants to subscribe", "user", client.UserID(), "client", client.ID(), "channel", e.Channel)
user, ok := livecontext.GetContextSignedUser(clientContextWithSpan)
if !ok {
user, err := identity.GetRequester(clientContextWithSpan)
if err != nil {
logger.Error("No user found in context", "user", client.UserID(), "client", client.ID(), "channel", e.Channel)
return centrifuge.SubscribeReply{}, centrifuge.ErrorInternal
}
@@ -830,8 +788,8 @@ func (g *GrafanaLive) handleOnSubscribe(clientContextWithSpan context.Context, c
func (g *GrafanaLive) handleOnPublish(clientCtxWithSpan context.Context, client *centrifuge.Client, e centrifuge.PublishEvent) (centrifuge.PublishReply, error) {
logger.Debug("Client wants to publish", "user", client.UserID(), "client", client.ID(), "channel", e.Channel)
user, ok := livecontext.GetContextSignedUser(clientCtxWithSpan)
if !ok {
user, err := identity.GetRequester(clientCtxWithSpan)
if err != nil {
logger.Error("No user found in context", "user", client.UserID(), "client", client.ID(), "channel", e.Channel)
return centrifuge.PublishReply{}, centrifuge.ErrorInternal
}
@@ -1083,7 +1041,7 @@ func (g *GrafanaLive) ClientCount(orgID int64, channel string) (int, error) {
}
func (g *GrafanaLive) HandleHTTPPublish(ctx *contextmodel.ReqContext) response.Response {
cmd := dtos.LivePublishCmd{}
cmd := model.LivePublishCmd{}
if err := web.Bind(ctx.Req, &cmd); err != nil {
return response.Error(http.StatusBadRequest, "bad request data", err)
}
@@ -1122,7 +1080,7 @@ func (g *GrafanaLive) HandleHTTPPublish(ctx *contextmodel.ReqContext) response.R
logger.Error("Error processing input", "user", user, "channel", channel, "error", err)
return response.Error(http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError), nil)
}
return response.JSON(http.StatusOK, dtos.LivePublishResponse{})
return response.JSON(http.StatusOK, model.LivePublishResponse{})
}
}
@@ -1150,7 +1108,7 @@ func (g *GrafanaLive) HandleHTTPPublish(ctx *contextmodel.ReqContext) response.R
}
}
logger.Debug("Publication successful", "identity", ctx.GetID(), "channel", cmd.Channel)
return response.JSON(http.StatusOK, dtos.LivePublishResponse{})
return response.JSON(http.StatusOK, model.LivePublishResponse{})
}
type streamChannelListResponse struct {
pkg/services/live/live_test.go
+12 -12
View File
@@ -11,20 +11,17 @@ import (
"testing"
"time"
"github.com/centrifugal/centrifuge"
"github.com/go-jose/go-jose/v4"
"github.com/go-jose/go-jose/v4/jwt"
"github.com/stretchr/testify/require"
"github.com/centrifugal/centrifuge"
"github.com/grafana/grafana/pkg/api/routing"
"github.com/grafana/grafana/pkg/apimachinery/identity"
"github.com/grafana/grafana/pkg/infra/db"
"github.com/grafana/grafana/pkg/infra/usagestats"
"github.com/grafana/grafana/pkg/services/accesscontrol/acimpl"
"github.com/grafana/grafana/pkg/services/dashboards"
"github.com/grafana/grafana/pkg/services/featuremgmt"
"github.com/grafana/grafana/pkg/services/live/livecontext"
"github.com/grafana/grafana/pkg/setting"
"github.com/grafana/grafana/pkg/tests/testsuite"
"github.com/grafana/grafana/pkg/util/testutil"
@@ -245,7 +242,7 @@ func Test_handleOnPublish_IDTokenExpiration(t *testing.T) {
t.Run("expired token", func(t *testing.T) {
expiration := time.Now().Add(-time.Hour)
token := createToken(t, &expiration)
ctx := livecontext.SetContextSignedUser(context.Background(), &identity.StaticRequester{IDToken: token})
ctx := identity.WithRequester(context.Background(), &identity.StaticRequester{IDToken: token})
reply, err := g.handleOnPublish(ctx, client, centrifuge.PublishEvent{
Channel: "test",
Data: []byte("test"),
@@ -257,7 +254,7 @@ func Test_handleOnPublish_IDTokenExpiration(t *testing.T) {
t.Run("unexpired token", func(t *testing.T) {
expiration := time.Now().Add(time.Hour)
token := createToken(t, &expiration)
ctx := livecontext.SetContextSignedUser(context.Background(), &identity.StaticRequester{IDToken: token})
ctx := identity.WithRequester(context.Background(), &identity.StaticRequester{IDToken: token})
reply, err := g.handleOnPublish(ctx, client, centrifuge.PublishEvent{
Channel: "test",
Data: []byte("test"),
@@ -280,7 +277,7 @@ func Test_handleOnRPC_IDTokenExpiration(t *testing.T) {
t.Run("expired token", func(t *testing.T) {
expiration := time.Now().Add(-time.Hour)
token := createToken(t, &expiration)
ctx := livecontext.SetContextSignedUser(context.Background(), &identity.StaticRequester{IDToken: token})
ctx := identity.WithRequester(context.Background(), &identity.StaticRequester{IDToken: token})
reply, err := g.handleOnRPC(ctx, client, centrifuge.RPCEvent{
Method: "grafana.query",
Data: []byte("test"),
@@ -292,7 +289,7 @@ func Test_handleOnRPC_IDTokenExpiration(t *testing.T) {
t.Run("unexpired token", func(t *testing.T) {
expiration := time.Now().Add(time.Hour)
token := createToken(t, &expiration)
ctx := livecontext.SetContextSignedUser(context.Background(), &identity.StaticRequester{IDToken: token})
ctx := identity.WithRequester(context.Background(), &identity.StaticRequester{IDToken: token})
reply, err := g.handleOnRPC(ctx, client, centrifuge.RPCEvent{
Method: "grafana.query",
Data: []byte("test"),
@@ -315,7 +312,7 @@ func Test_handleOnSubscribe_IDTokenExpiration(t *testing.T) {
t.Run("expired token", func(t *testing.T) {
expiration := time.Now().Add(-time.Hour)
token := createToken(t, &expiration)
ctx := livecontext.SetContextSignedUser(context.Background(), &identity.StaticRequester{IDToken: token})
ctx := identity.WithRequester(context.Background(), &identity.StaticRequester{IDToken: token})
reply, err := g.handleOnSubscribe(ctx, client, centrifuge.SubscribeEvent{
Channel: "test",
})
@@ -326,7 +323,7 @@ func Test_handleOnSubscribe_IDTokenExpiration(t *testing.T) {
t.Run("unexpired token", func(t *testing.T) {
expiration := time.Now().Add(time.Hour)
token := createToken(t, &expiration)
ctx := livecontext.SetContextSignedUser(context.Background(), &identity.StaticRequester{IDToken: token})
ctx := identity.WithRequester(context.Background(), &identity.StaticRequester{IDToken: token})
reply, err := g.handleOnSubscribe(ctx, client, centrifuge.SubscribeEvent{
Channel: "test",
})
@@ -347,10 +344,8 @@ func setupLiveService(cfg *setting.Cfg, t *testing.T) (*GrafanaLive, error) {
cfg,
routing.NewRouteRegister(),
nil, nil, nil, nil,
db.InitTestDB(t),
nil,
&usagestats.UsageStatsMock{T: t},
nil,
featuremgmt.WithFeatures(),
acimpl.ProvideAccessControl(featuremgmt.WithFeatures()),
&dashboards.FakeDashboardService{},
@@ -361,7 +356,12 @@ type dummyTransport struct {
name string
}
var (
_ centrifuge.Transport = (*dummyTransport)(nil)
)
func (t *dummyTransport) Name() string { return t.name }
func (t *dummyTransport) AcceptProtocol() string { return "" }
func (t *dummyTransport) Protocol() centrifuge.ProtocolType { return centrifuge.ProtocolTypeJSON }
func (t *dummyTransport) ProtocolVersion() centrifuge.ProtocolVersion {
return centrifuge.ProtocolVersion2
pkg/services/live/livecontext/context.go
-19
View File
@@ -2,27 +2,8 @@ package livecontext
import (
"context"
"github.com/grafana/grafana/pkg/apimachinery/identity"
)
type signedUserContextKeyType int
var signedUserContextKey signedUserContextKeyType
func SetContextSignedUser(ctx context.Context, user identity.Requester) context.Context {
ctx = context.WithValue(ctx, signedUserContextKey, user)
return ctx
}
func GetContextSignedUser(ctx context.Context) (identity.Requester, bool) {
if val := ctx.Value(signedUserContextKey); val != nil {
user, ok := val.(identity.Requester)
return user, ok
}
return nil, false
}
type streamIDContextKey struct{}
func SetContextStreamID(ctx context.Context, streamID string) context.Context {
pkg/services/live/model/model.go
+4 -16
View File
@@ -67,21 +67,9 @@ type ChannelHandlerFactory interface {
GetHandlerForPath(path string) (ChannelHandler, error)
}
type LiveMessage struct {
ID int64 `xorm:"pk autoincr 'id'"`
OrgID int64 `xorm:"org_id"`
Channel string
Data json.RawMessage
Published time.Time
type LivePublishCmd struct {
Channel string `json:"channel"`
Data json.RawMessage `json:"data,omitempty"`
}
type SaveLiveMessageQuery struct {
OrgID int64 `xorm:"org_id"`
Channel string
Data json.RawMessage
}
type GetLiveMessageQuery struct {
OrgID int64 `xorm:"org_id"`
Channel string
}
type LivePublishResponse struct{}
pkg/services/live/pipeline/data_output_builtin.go
+3 -3
View File
@@ -6,7 +6,7 @@ import (
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana/pkg/services/live/livecontext"
"github.com/grafana/grafana/pkg/apimachinery/identity"
"github.com/grafana/grafana/pkg/services/live/model"
)
@@ -25,8 +25,8 @@ func (s *BuiltinDataOutput) Type() string {
}
func (s *BuiltinDataOutput) OutputData(ctx context.Context, vars Vars, data []byte) ([]*ChannelData, error) {
u, ok := livecontext.GetContextSignedUser(ctx)
if !ok {
u, err := identity.GetRequester(ctx)
if err != nil {
return nil, errors.New("user not found in context")
}
handler, _, err := s.channelHandlerGetter.GetChannelHandler(ctx, u, vars.Channel)
pkg/services/live/pipeline/subscribe_builtin.go
+2 -3
View File
@@ -7,7 +7,6 @@ import (
"github.com/grafana/grafana-plugin-sdk-go/live"
"github.com/grafana/grafana/pkg/apimachinery/identity"
"github.com/grafana/grafana/pkg/services/live/livecontext"
"github.com/grafana/grafana/pkg/services/live/model"
)
@@ -30,8 +29,8 @@ func (s *BuiltinSubscriber) Type() string {
}
func (s *BuiltinSubscriber) Subscribe(ctx context.Context, vars Vars, data []byte) (model.SubscribeReply, backend.SubscribeStreamStatus, error) {
u, ok := livecontext.GetContextSignedUser(ctx)
if !ok {
u, err := identity.GetRequester(ctx)
if err != nil {
return model.SubscribeReply{}, backend.SubscribeStreamStatusPermissionDenied, nil
}
handler, _, err := s.channelHandlerGetter.GetChannelHandler(ctx, u, vars.Channel)
pkg/services/live/pipeline/subscribe_managed_stream.go
+3 -3
View File
@@ -5,7 +5,7 @@ import (
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana/pkg/services/live/livecontext"
"github.com/grafana/grafana/pkg/apimachinery/identity"
"github.com/grafana/grafana/pkg/services/live/managedstream"
"github.com/grafana/grafana/pkg/services/live/model"
)
@@ -30,8 +30,8 @@ func (s *ManagedStreamSubscriber) Subscribe(ctx context.Context, vars Vars, _ []
logger.Error("Error getting managed stream", "error", err)
return model.SubscribeReply{}, 0, err
}
u, ok := livecontext.GetContextSignedUser(ctx)
if !ok {
u, err := identity.GetRequester(ctx)
if err != nil {
return model.SubscribeReply{}, backend.SubscribeStreamStatusPermissionDenied, nil
}
return stream.OnSubscribe(ctx, u, model.SubscribeEvent{
pkg/services/live/pipeline/subscribe_multiple.go
-1
View File
@@ -4,7 +4,6 @@ import (
"context"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana/pkg/services/live/model"
)
pkg/services/live/pushws/push_pipeline.go
+3 -2
View File
@@ -5,6 +5,7 @@ import (
"github.com/gorilla/websocket"
"github.com/grafana/grafana/pkg/apimachinery/identity"
"github.com/grafana/grafana/pkg/services/live/convert"
"github.com/grafana/grafana/pkg/services/live/livecontext"
"github.com/grafana/grafana/pkg/services/live/pipeline"
@@ -44,8 +45,8 @@ func (s *PipelinePushHandler) ServeHTTP(rw http.ResponseWriter, r *http.Request)
return
}
user, ok := livecontext.GetContextSignedUser(r.Context())
if !ok {
user, err := identity.GetRequester(r.Context())
if err != nil {
logger.Error("No user found in context")
rw.WriteHeader(http.StatusInternalServerError)
return
pkg/services/live/pushws/push_stream.go
+4 -3
View File
@@ -5,8 +5,9 @@ import (
"time"
"github.com/gorilla/websocket"
liveDto "github.com/grafana/grafana-plugin-sdk-go/live"
liveDto "github.com/grafana/grafana-plugin-sdk-go/live"
"github.com/grafana/grafana/pkg/apimachinery/identity"
"github.com/grafana/grafana/pkg/services/live/convert"
"github.com/grafana/grafana/pkg/services/live/livecontext"
"github.com/grafana/grafana/pkg/services/live/managedstream"
@@ -47,8 +48,8 @@ func (s *Handler) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
return
}
user, ok := livecontext.GetContextSignedUser(r.Context())
if !ok {
user, err := identity.GetRequester(r.Context())
if err != nil {
logger.Error("No user found in context")
rw.WriteHeader(http.StatusInternalServerError)
return