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

Live: Move dashboard events from the raw http server to the apiserver (#115066)

Browse Source
This commit is contained in:
Ryan McKinley
2025-12-11 09:26:35 +03:00
committed by GitHub
parent a6043deb33
commit 8542b2f6a2
8 changed files with 66 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/api/dashboard.go
-32
View File
@@ -380,13 +380,6 @@ func (hs *HTTPServer) deleteDashboard(c *contextmodel.ReqContext) response.Respo
return dashboardErrResponse(err, "Failed to delete dashboard")
}
if hs.Live != nil {
err := hs.Live.GrafanaScope.Dashboards.DashboardDeleted(c.GetOrgID(), c.SignedInUser, dash.UID)
if err != nil {
hs.log.Error("Failed to broadcast delete info", "dashboard", dash.UID, "error", err)
}
}
return response.JSON(http.StatusOK, util.DynMap{
"title": dash.Title,
"message": fmt.Sprintf("Dashboard %s deleted", dash.Title),
@@ -482,31 +475,6 @@ func (hs *HTTPServer) postDashboard(c *contextmodel.ReqContext, cmd dashboards.S
}
dashboard, saveErr := hs.DashboardService.SaveDashboard(ctx, dashItem, allowUiUpdate)
if hs.Live != nil {
// Tell everyone listening that the dashboard changed
if dashboard == nil {
dashboard = dash // the original request
}
// This will broadcast all save requests only if a `gitops` observer exists.
// gitops is useful when trying to save dashboards in an environment where the user can not save
channel := hs.Live.GrafanaScope.Dashboards
liveerr := channel.DashboardSaved(c.GetOrgID(), c.SignedInUser, cmd.Message, dashboard, saveErr)
// When an error exists, but the value broadcast to a gitops listener return 202
if liveerr == nil && saveErr != nil && channel.HasGitOpsObserver(c.GetOrgID()) {
return response.JSON(http.StatusAccepted, util.DynMap{
"status": "pending",
"message": "changes were broadcast to the gitops listener",
})
}
if liveerr != nil {
hs.log.Warn("Unable to broadcast save event", "uid", dashboard.UID, "error", liveerr)
}
}
if saveErr != nil {
return apierrors.ToDashboardErrorResponse(ctx, hs.pluginStore, saveErr)
}
pkg/registry/apis/dashboard/dashboard_storage.go
+33 -7
View File
@@ -7,20 +7,41 @@ import (
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apiserver/pkg/registry/rest"
"github.com/grafana/grafana-app-sdk/logging"
grafanarest "github.com/grafana/grafana/pkg/apiserver/rest"
"github.com/grafana/grafana/pkg/services/accesscontrol"
"github.com/grafana/grafana/pkg/services/apiserver/endpoints/request"
"github.com/grafana/grafana/pkg/services/live"
)
// dashboardStoragePermissionWrapper is a wrapper around the grafanarest.Storage that adds dashboard permissions handling
// when dual writing is enabled.
type dashboardStoragePermissionWrapper struct {
dashboardPermissionsSvc accesscontrol.DashboardPermissionsService
// dashboardStorageWrapper is a wrapper around the grafanarest.Storage so it will:
// 1. support adds dashboard permissions handling
// 2. broadcast changes to grafana live
// when running in single tenant mode
type dashboardStorageWrapper struct {
grafanarest.Storage
dashboardPermissionsSvc accesscontrol.DashboardPermissionsService
live live.DashboardActivityChannel
}
func (d dashboardStoragePermissionWrapper) Delete(ctx context.Context, name string, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions) (runtime.Object, bool, error) {
info, err := request.NamespaceInfoFrom(ctx, true)
func (d dashboardStorageWrapper) Update(ctx context.Context, name string, objInfo rest.UpdatedObjectInfo, createValidation rest.ValidateObjectFunc, updateValidation rest.ValidateObjectUpdateFunc, forceAllowCreate bool, options *metav1.UpdateOptions) (runtime.Object, bool, error) {
ns, err := request.NamespaceInfoFrom(ctx, true)
if err != nil {
return nil, false, err
}
obj, created, err := d.Storage.Update(ctx, name, objInfo, createValidation, updateValidation, forceAllowCreate, options)
if err == nil && ns.OrgID > 0 && d.live != nil {
if err := d.live.DashboardSaved(ns.OrgID, name); err != nil {
logging.FromContext(ctx).Info("live dashboard update failed", "err", err)
}
}
return obj, created, err
}
func (d dashboardStorageWrapper) Delete(ctx context.Context, name string, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions) (runtime.Object, bool, error) {
ns, err := request.NamespaceInfoFrom(ctx, true)
if err != nil {
return nil, false, err
}
@@ -28,7 +49,12 @@ func (d dashboardStoragePermissionWrapper) Delete(ctx context.Context, name stri
if err != nil {
return obj, async, err
}
if accessErr := d.dashboardPermissionsSvc.DeleteResourcePermissions(ctx, info.OrgID, name); accessErr != nil {
if ns.OrgID > 0 && d.live != nil {
if err := d.live.DashboardDeleted(ns.OrgID, name); err != nil {
logging.FromContext(ctx).Info("live dashboard update failed", "err", err)
}
}
if accessErr := d.dashboardPermissionsSvc.DeleteResourcePermissions(ctx, ns.OrgID, name); accessErr != nil {
return obj, async, accessErr
}
return obj, async, nil
pkg/registry/apis/dashboard/register.go
+7 -2
View File
@@ -56,6 +56,7 @@ import (
"github.com/grafana/grafana/pkg/services/featuremgmt"
"github.com/grafana/grafana/pkg/services/libraryelements"
"github.com/grafana/grafana/pkg/services/librarypanels"
"github.com/grafana/grafana/pkg/services/live"
"github.com/grafana/grafana/pkg/services/provisioning"
"github.com/grafana/grafana/pkg/services/publicdashboards"
"github.com/grafana/grafana/pkg/services/quota"
@@ -122,6 +123,7 @@ type DashboardsAPIBuilder struct {
snapshotService dashboardsnapshots.Service
snapshotOptions dashv0.SnapshotSharingOptions
namespacer request.NamespaceMapper
dashboardActivityChannel live.DashboardActivityChannel
isStandalone bool // skips any handling including anything to do with legacy storage
}
@@ -151,6 +153,7 @@ func RegisterAPIService(
libraryPanels libraryelements.Service,
publicDashboardService publicdashboards.Service,
snapshotService dashboardsnapshots.Service,
dashboardActivityChannel live.DashboardActivityChannel,
) *DashboardsAPIBuilder {
dbp := legacysql.NewDatabaseProvider(sql)
namespacer := request.GetNamespaceMapper(cfg)
@@ -185,6 +188,7 @@ func RegisterAPIService(
snapshotService: snapshotService,
snapshotOptions: snapshotOptions,
namespacer: namespacer,
dashboardActivityChannel: dashboardActivityChannel,
legacy: &DashboardStorage{
Access: legacy.NewDashboardSQLAccess(dbp, namespacer, dashStore, provisioning, libraryPanelSvc, sorter, dashboardPermissionsSvc, accessControl, features),
DashboardService: dashboardService,
@@ -696,9 +700,10 @@ func (b *DashboardsAPIBuilder) storageForVersion(
if err != nil {
return err
}
storage[dashboards.StoragePath()] = dashboardStoragePermissionWrapper{
dashboardPermissionsSvc: b.dashboardPermissionsSvc,
storage[dashboards.StoragePath()] = dashboardStorageWrapper{
Storage: dw,
dashboardPermissionsSvc: b.dashboardPermissionsSvc,
live: b.dashboardActivityChannel,
}
// Register the DTO endpoint that will consolidate all dashboard bits
pkg/server/wire.go
+1
View File
@@ -280,6 +280,7 @@ var wireBasicSet = wire.NewSet(
store.ProvideService,
store.ProvideSystemUsersService,
live.ProvideService,
live.ProvideDashboardActivityChannel,
pushhttp.ProvideService,
contexthandler.ProvideService,
ldapservice.ProvideService,
pkg/server/wire_gen.go
Generated
+5 -3
View File
File diff suppressed because one or more lines are too long
pkg/services/live/features/dashboard.go
+12 -75
View File
@@ -22,39 +22,13 @@ const (
ActionDeleted actionType = "deleted"
EditingStarted actionType = "editing-started"
//EditingFinished actionType = "editing-finished"
GitopsChannel = "grafana/dashboard/gitops"
)
type userDisplayDTO struct {
ID int64 `json:"id,omitempty"`
UID string `json:"uid,omitempty"`
Name string `json:"name,omitempty"`
Login string `json:"login,omitempty"`
AvatarURL string `json:"avatarUrl"`
}
// Static function to parse a requester into a userDisplayDTO
func newUserDisplayDTOFromRequester(requester identity.Requester) *userDisplayDTO {
// nolint:staticcheck
userID, _ := requester.GetInternalID()
return &userDisplayDTO{
ID: userID,
UID: requester.GetRawIdentifier(),
Login: requester.GetLogin(),
Name: requester.GetName(),
}
}
// DashboardEvent events related to dashboards
type dashboardEvent struct {
UID string `json:"uid"`
Action actionType `json:"action"` // saved, editing, deleted
User *userDisplayDTO `json:"user,omitempty"`
SessionID string `json:"sessionId,omitempty"`
Message string `json:"message,omitempty"`
Dashboard *dashboards.Dashboard `json:"dashboard,omitempty"`
Error string `json:"error,omitempty"`
UID string `json:"uid"`
Action actionType `json:"action"` // saved, editing, deleted
SessionID string `json:"sessionId,omitempty"`
}
// DashboardHandler manages all the `grafana/dashboard/*` channels
@@ -134,9 +108,6 @@ func (h *DashboardHandler) OnPublish(ctx context.Context, requester identity.Req
return model.PublishReply{}, backend.PublishStreamStatusNotFound, nil // NOOP
}
// Tell everyone who is editing
event.User = newUserDisplayDTOFromRequester(requester)
msg, err := json.Marshal(event)
if err != nil {
return model.PublishReply{}, backend.PublishStreamStatusNotFound, fmt.Errorf("internal error")
@@ -153,55 +124,21 @@ func (h *DashboardHandler) publish(orgID int64, event dashboardEvent) error {
if err != nil {
return err
}
// Only broadcast non-error events
if event.Error == "" {
err = h.Publisher(orgID, "grafana/dashboard/uid/"+event.UID, msg)
if err != nil {
return err
}
}
// Send everything to the gitops channel
return h.Publisher(orgID, GitopsChannel, msg)
return h.Publisher(orgID, "grafana/dashboard/uid/"+event.UID, msg)
}
// DashboardSaved will broadcast to all connected dashboards
func (h *DashboardHandler) DashboardSaved(orgID int64, requester identity.Requester, message string, dashboard *dashboards.Dashboard, err error) error {
if err != nil && !h.HasGitOpsObserver(orgID) {
return nil // only broadcast if it was OK
}
msg := dashboardEvent{
UID: dashboard.UID,
Action: ActionSaved,
User: newUserDisplayDTOFromRequester(requester),
Message: message,
Dashboard: dashboard,
}
if err != nil {
msg.Error = err.Error()
}
return h.publish(orgID, msg)
}
// DashboardDeleted will broadcast to all connected dashboards
func (h *DashboardHandler) DashboardDeleted(orgID int64, requester identity.Requester, uid string) error {
func (h *DashboardHandler) DashboardSaved(orgID int64, uid string) error {
return h.publish(orgID, dashboardEvent{
UID: uid,
Action: ActionDeleted,
User: newUserDisplayDTOFromRequester(requester),
Action: ActionSaved,
})
}
// HasGitOpsObserver will return true if anyone is listening to the `gitops` channel
func (h *DashboardHandler) HasGitOpsObserver(orgID int64) bool {
count, err := h.ClientCount(orgID, GitopsChannel)
if err != nil {
logger.Error("Error getting client count", "error", err)
return false
}
return count > 0
// DashboardDeleted will broadcast to all connected dashboards
func (h *DashboardHandler) DashboardDeleted(orgID int64, uid string) error {
return h.publish(orgID, dashboardEvent{
UID: uid,
Action: ActionDeleted,
})
}
pkg/services/live/live.go
+8 -15
View File
@@ -513,18 +513,17 @@ type GrafanaLive struct {
// DashboardActivityChannel is a service to advertise dashboard activity
type DashboardActivityChannel interface {
// Called when a dashboard is saved -- this includes the error so we can support a
// gitops workflow that knows if the value was saved to the local database or not
// in many cases all direct save requests will fail, but the request should be forwarded
// to any gitops observers
DashboardSaved(orgID int64, requester identity.Requester, message string, dashboard *dashboards.Dashboard, err error) error
// Called when a dashboard is saved
DashboardSaved(orgID int64, uid string) error
// Called when a dashboard is deleted
DashboardDeleted(orgID int64, requester identity.Requester, uid string) error
DashboardDeleted(orgID int64, uid string) error
}
// Experimental! Indicate is GitOps is active. This really means
// someone is subscribed to the `grafana/dashboards/gitops` channel
HasGitOpsObserver(orgID int64) bool
// ProvideDashboardActivityChannel extracts the DashboardActivityChannel from GrafanaLive.
// This is used by wire to inject the channel into the dashboard API service.
func ProvideDashboardActivityChannel(live *GrafanaLive) DashboardActivityChannel {
return live.GrafanaScope.Dashboards
}
func (g *GrafanaLive) getStreamPlugin(ctx context.Context, pluginID string) (backend.StreamHandler, error) {
@@ -1178,12 +1177,6 @@ func (g *GrafanaLive) HandleListHTTP(c *contextmodel.ReqContext) response.Respon
// HandleInfoHTTP special http response for
func (g *GrafanaLive) HandleInfoHTTP(ctx *contextmodel.ReqContext) response.Response {
path := web.Params(ctx.Req)["*"]
if path == "grafana/dashboards/gitops" {
return response.JSON(http.StatusOK, util.DynMap{
"active": g.GrafanaScope.Dashboards.HasGitOpsObserver(ctx.GetOrgID()),
})
}
return response.JSONStreaming(http.StatusNotFound, util.DynMap{
"message": "Info is not supported for this channel",
})
public/app/features/live/dashboard/types.ts
-1
View File
@@ -8,7 +8,6 @@ export enum DashboardEventAction {
export interface DashboardEvent {
uid: string;
action: DashboardEventAction;
userId?: number;
message?: string;
sessionId?: string;
timestamp?: number;