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

Provisioning: Merge watch support into live (#102618)

Browse Source
This commit is contained in:
Ryan McKinley
2025-03-21 16:45:25 +02:00
committed by GitHub
parent 77c5e0eeb2
commit 1a00801e6a
18 changed files with 723 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/services/live/features/watch.go
+214
View File
@@ -0,0 +1,214 @@
package features
import (
"context"
"fmt"
"strings"
"sync"
v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/watch"
"k8s.io/client-go/dynamic"
provisioning "github.com/grafana/grafana/pkg/apis/provisioning/v0alpha1"
"github.com/grafana/authlib/types"
"github.com/grafana/grafana-app-sdk/logging"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana-plugin-sdk-go/data/utils/jsoniter"
"github.com/grafana/grafana-plugin-sdk-go/live"
"github.com/grafana/grafana/pkg/apimachinery/identity"
"github.com/grafana/grafana/pkg/services/apiserver"
"github.com/grafana/grafana/pkg/services/live/model"
)
// WatchRunner will start a watch task and broadcast results
type WatchRunner struct {
publisher model.ChannelPublisher
configProvider apiserver.RestConfigProvider
watchingMu sync.Mutex
watching map[string]*watcher
}
func NewWatchRunner(publisher model.ChannelPublisher, configProvider apiserver.RestConfigProvider) *WatchRunner {
return &WatchRunner{
publisher: publisher,
configProvider: configProvider,
watching: make(map[string]*watcher),
}
}
func (b *WatchRunner) GetHandlerForPath(_ string) (model.ChannelHandler, error) {
return b, nil // all dashboards share the same handler
}
// Valid paths look like: {version}/{resource}[={name}]/{user.uid}
// * v0alpha1/dashboards/u12345
// * v0alpha1/dashboards=ABCD/u12345
func (b *WatchRunner) OnSubscribe(ctx context.Context, u identity.Requester, e model.SubscribeEvent) (model.SubscribeReply, backend.SubscribeStreamStatus, error) {
// To make sure we do not share resources across users, in clude the UID in the path
userID := u.GetIdentifier()
if userID == "" {
return model.SubscribeReply{}, backend.SubscribeStreamStatusPermissionDenied, fmt.Errorf("missing user identity")
}
if !strings.HasSuffix(e.Path, userID) {
return model.SubscribeReply{}, backend.SubscribeStreamStatusPermissionDenied, fmt.Errorf("path must end with user uid (%s)", userID)
}
// While testing with provisioning repositories, we will limit this to admin only
if !u.HasRole(identity.RoleAdmin) {
return model.SubscribeReply{}, backend.SubscribeStreamStatusPermissionDenied, fmt.Errorf("only admin users for now")
}
b.watchingMu.Lock()
defer b.watchingMu.Unlock()
current, ok := b.watching[e.Channel]
if ok && !current.done {
return model.SubscribeReply{
JoinLeave: false,
Presence: false,
Recover: false,
}, backend.SubscribeStreamStatusOK, nil
}
// Try to start a watcher for this request
gvr, name, err := parseWatchRequest(e.Channel, userID)
if err != nil {
return model.SubscribeReply{}, backend.SubscribeStreamStatusNotFound, err
}
// Test this with only provisiong support -- then we can evaluate a broader rollout
if gvr.Group != provisioning.GROUP {
return model.SubscribeReply{}, backend.SubscribeStreamStatusPermissionDenied,
fmt.Errorf("watching provisioned resources is OK allowed (for now)")
}
requester := types.WithAuthInfo(context.Background(), u)
cfg, err := b.configProvider.GetRestConfig(requester)
if err != nil {
return model.SubscribeReply{}, backend.SubscribeStreamStatusNotFound, err
}
uclient, err := dynamic.NewForConfig(cfg)
if err != nil {
return model.SubscribeReply{}, backend.SubscribeStreamStatusNotFound, err
}
client := uclient.Resource(gvr).Namespace(u.GetNamespace())
opts := v1.ListOptions{}
if len(name) > 1 {
opts.FieldSelector = "metadata.name=" + name
}
watch, err := client.Watch(requester, opts)
if err != nil {
return model.SubscribeReply{}, backend.SubscribeStreamStatusNotFound, err
}
current = &watcher{
orgId: u.GetOrgID(),
channel: e.Channel,
publisher: b.publisher,
watch: watch,
}
b.watching[e.Channel] = current
go current.run(ctx)
return model.SubscribeReply{
JoinLeave: false, // need unsubscribe envents
Presence: false,
Recover: false,
}, backend.SubscribeStreamStatusOK, nil
}
func parseWatchRequest(channel string, user string) (gvr schema.GroupVersionResource, name string, err error) {
addr, err := live.ParseChannel(channel)
if err != nil {
return gvr, "", err
}
parts := strings.Split(addr.Path, "/")
if len(parts) != 3 {
return gvr, "", fmt.Errorf("expecting path: {version}/{resource}={name}/{user}")
}
if parts[2] != user {
return gvr, "", fmt.Errorf("expecting user suffix: %s", user)
}
resource := strings.Split(parts[1], "=")
gvr = schema.GroupVersionResource{
Group: addr.Namespace,
Version: parts[0],
Resource: resource[0],
}
if len(resource) > 1 {
name = resource[1]
}
return gvr, name, nil
}
// OnPublish is called when a client wants to broadcast on the websocket
func (b *WatchRunner) OnPublish(_ context.Context, u identity.Requester, e model.PublishEvent) (model.PublishReply, backend.PublishStreamStatus, error) {
return model.PublishReply{}, backend.PublishStreamStatusNotFound, fmt.Errorf("watch does not support publish")
}
type watcher struct {
orgId int64
channel string
publisher model.ChannelPublisher
done bool
watch watch.Interface
}
func (b *watcher) run(ctx context.Context) {
logger := logging.FromContext(ctx).With("channel", b.channel)
ch := b.watch.ResultChan()
for {
select {
// This is sent when there are no longer any subscriptions
case <-ctx.Done():
logger.Info("context done", "channel", b.channel)
b.watch.Stop()
b.done = true
return
// Each watch event
case event, ok := <-ch:
if !ok {
logger.Info("watch stream broken", "channel", b.channel)
b.watch.Stop()
b.done = true // will force reconnect from the frontend
return
}
cfg := jsoniter.ConfigCompatibleWithStandardLibrary
stream := cfg.BorrowStream(nil)
defer cfg.ReturnStream(stream)
// regular json.Marshal() uses upper case
stream.WriteObjectStart()
stream.WriteObjectField("type")
stream.WriteString(string(event.Type))
stream.WriteMore()
stream.WriteObjectField("object")
stream.WriteVal(event.Object)
stream.WriteObjectEnd()
buf := stream.Buffer()
data := make([]byte, len(buf))
copy(data, buf)
err := b.publisher(b.orgId, b.channel, data)
if err != nil {
logger.Error("publish error", "channel", b.channel, "err", err)
b.watch.Stop()
b.done = true // will force reconnect from the frontend
continue
}
}
}
}
pkg/services/live/live.go
+16 -1
View File
@@ -36,6 +36,7 @@ import (
"github.com/grafana/grafana/pkg/plugins"
"github.com/grafana/grafana/pkg/services/accesscontrol"
"github.com/grafana/grafana/pkg/services/annotations"
"github.com/grafana/grafana/pkg/services/apiserver"
contextmodel "github.com/grafana/grafana/pkg/services/contexthandler/model"
"github.com/grafana/grafana/pkg/services/dashboards"
"github.com/grafana/grafana/pkg/services/datasources"
@@ -79,7 +80,7 @@ func ProvideService(plugCtxProvider *plugincontext.Provider, cfg *setting.Cfg, r
dataSourceCache datasources.CacheService, sqlStore db.DB, secretsService secrets.Service,
usageStatsService usagestats.Service, queryDataService query.Service, toggles featuremgmt.FeatureToggles,
accessControl accesscontrol.AccessControl, dashboardService dashboards.DashboardService, annotationsRepo annotations.Repository,
orgService org.Service) (*GrafanaLive, error) {
orgService org.Service, configProvider apiserver.RestConfigProvider) (*GrafanaLive, error) {
g := &GrafanaLive{
Cfg: cfg,
Features: toggles,
@@ -191,6 +192,11 @@ func ProvideService(plugCtxProvider *plugincontext.Provider, cfg *setting.Cfg, r
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
if toggles.IsEnabledGlobally(featuremgmt.FlagProvisioning) {
g.GrafanaScope.Features["watch"] = features.NewWatchRunner(g.Publish, configProvider)
}
g.surveyCaller = survey.NewCaller(managedStreamRunner, node)
err = g.surveyCaller.SetupHandlers()
if err != nil {
@@ -889,6 +895,8 @@ func (g *GrafanaLive) GetChannelHandlerFactory(ctx context.Context, user identit
switch scope {
case live.ScopeGrafana:
return g.handleGrafanaScope(user, namespace)
case "watch": // TODO: live.ScopeWatch: update 275 https://github.com/grafana/grafana-plugin-sdk-go/releases
return g.handleWatchScope()
case live.ScopePlugin:
return g.handlePluginScope(ctx, user, namespace)
case live.ScopeDatasource:
@@ -907,6 +915,13 @@ func (g *GrafanaLive) handleGrafanaScope(_ identity.Requester, namespace string)
return nil, fmt.Errorf("unknown feature: %q", namespace)
}
func (g *GrafanaLive) handleWatchScope() (model.ChannelHandlerFactory, error) {
if p, ok := g.GrafanaScope.Features["watch"]; ok {
return p, nil
}
return nil, fmt.Errorf("watch not registered")
}
func (g *GrafanaLive) handlePluginScope(ctx context.Context, _ identity.Requester, namespace string) (model.ChannelHandlerFactory, error) {
streamHandler, err := g.getStreamPlugin(ctx, namespace)
if err != nil {
pkg/services/live/live_test.go
+5 -1
View File
@@ -36,7 +36,11 @@ func Test_provideLiveService_RedisUnavailable(t *testing.T) {
nil,
&usagestats.UsageStatsMock{T: t},
nil,
featuremgmt.WithFeatures(), acimpl.ProvideAccessControl(featuremgmt.WithFeatures()), &dashboards.FakeDashboardService{}, annotationstest.NewFakeAnnotationsRepo(), nil)
featuremgmt.WithFeatures(),
acimpl.ProvideAccessControl(featuremgmt.WithFeatures()),
&dashboards.FakeDashboardService{},
annotationstest.NewFakeAnnotationsRepo(),
nil, nil)
// Proceeds without live HA if redis is unavaialble
require.NoError(t, err)