From 9fce4311e9bbce15540ba43f43e93cef5981521e Mon Sep 17 00:00:00 2001 From: Ben Sully Date: Thu, 20 Mar 2025 16:50:52 +0000 Subject: [PATCH] Live: allow publishing over Centrifuge subscription (#102325) * Live: allow publishing over Centrifuge subscription Currently when publishing over a Grafana Live channel, the data is sent over the HTTP API. This works fine when there is only a single Grafana instance running, but when there are multiple instances, the data will only hit one instance, which is often not desired: sometimes you need to guarantee that the data appears on the same instance that the frontend is connected to. An example of this is in the Grafana LLM app when running the MCP server. The MCP protocol is stateful; users subscribe to a channel to get a long-lived stream of server-sent events, then send subsequent requests to the server to get further results. If there are multiple Grafana instances running then the requests are likely to land on an instance other than the one that the user is connected to. This commit adds a new option to the `GrafanaLiveSrv` interface that allows the user to publish data over the Centrifuge subscription instead of the HTTP API. This is not the default and should rarely be used, but is required to fulfil certain use cases. * Address nits from code review Co-authored-by: kay delaney <45561153+kaydelaney@users.noreply.github.com> --------- Co-authored-by: kay delaney <45561153+kaydelaney@users.noreply.github.com> --- packages/grafana-runtime/src/services/live.ts | 16 +++++++++++++++- public/app/features/live/centrifuge/channel.ts | 2 ++ public/app/features/live/centrifuge/service.ts | 10 +++++++++- .../features/live/centrifuge/service.worker.ts | 6 +++++- .../live/centrifuge/serviceWorkerProxy.ts | 4 ++++ public/app/features/live/live.ts | 6 +++++- 6 files changed, 40 insertions(+), 4 deletions(-) diff --git a/packages/grafana-runtime/src/services/live.ts b/packages/grafana-runtime/src/services/live.ts index e3787383995..41c05ac76bd 100644 --- a/packages/grafana-runtime/src/services/live.ts +++ b/packages/grafana-runtime/src/services/live.ts @@ -39,6 +39,20 @@ export interface LiveQueryDataOptions { body: unknown; // processed queries, same as sent to `/api/query/ds` } +/** + * @alpha -- experimental + */ +export interface LivePublishOptions { + /** + * Publish the data over the websocket instead of the HTTP API. + * + * This is not recommended for most use cases. + * + * @experimental + */ + useSocket?: boolean; +} + /** * @alpha -- experimental */ @@ -79,7 +93,7 @@ export interface GrafanaLiveSrv { * * @alpha -- experimental */ - publish(address: LiveChannelAddress, data: unknown): Promise; + publish(address: LiveChannelAddress, data: unknown, options?: LivePublishOptions): Promise; } let singletonInstance: GrafanaLiveSrv; diff --git a/public/app/features/live/centrifuge/channel.ts b/public/app/features/live/centrifuge/channel.ts index fd2f6a109f1..2edb1fc470b 100644 --- a/public/app/features/live/centrifuge/channel.ts +++ b/public/app/features/live/centrifuge/channel.ts @@ -175,6 +175,8 @@ export class CentrifugeLiveChannel { }); } + publish = (data: unknown) => this.subscription?.publish(data); + /** * This will close and terminate all streams for this channel */ diff --git a/public/app/features/live/centrifuge/service.ts b/public/app/features/live/centrifuge/service.ts index 8fb666a604c..b99a78d2f10 100644 --- a/public/app/features/live/centrifuge/service.ts +++ b/public/app/features/live/centrifuge/service.ts @@ -20,6 +20,7 @@ import { FetchResponse } from '@grafana/runtime/src/services/backendSrv'; import { GrafanaLiveSrv, LiveDataStreamOptions, + LivePublishOptions, LiveQueryDataOptions, StreamingFrameAction, StreamingFrameOptions, @@ -42,7 +43,7 @@ export type CentrifugeSrvDeps = { export type StreamingDataQueryResponse = Omit & { data: [StreamingResponseData] }; -export type CentrifugeSrv = Omit & { +export type CentrifugeSrv = Omit & { getDataStream: (options: LiveDataStreamOptions) => Observable; getQueryData: ( options: LiveQueryDataOptions @@ -244,6 +245,13 @@ export class CentrifugeService implements CentrifugeSrv { getPresence: CentrifugeSrv['getPresence'] = (address) => { return this.getChannel(address).getPresence(); }; + + /** + * Publish into a channel. + */ + publish = async (address: LiveChannelAddress, data: unknown, options?: LivePublishOptions) => { + return this.getChannel(address).publish(data); + }; } // This is used to give a unique key for each stream. The actual value does not matter diff --git a/public/app/features/live/centrifuge/service.worker.ts b/public/app/features/live/centrifuge/service.worker.ts index a168bc34824..0c535bc9cb8 100644 --- a/public/app/features/live/centrifuge/service.worker.ts +++ b/public/app/features/live/centrifuge/service.worker.ts @@ -3,7 +3,7 @@ import './transferHandlers'; import * as comlink from 'comlink'; import { LiveChannelAddress } from '@grafana/data'; -import { LiveDataStreamOptions, LiveQueryDataOptions } from '@grafana/runtime'; +import { LiveDataStreamOptions, LivePublishOptions, LiveQueryDataOptions } from '@grafana/runtime'; import { remoteObservableAsObservable } from './remoteObservable'; import { CentrifugeService, CentrifugeSrvDeps } from './service'; @@ -42,6 +42,9 @@ const getPresence = async (address: LiveChannelAddress) => { return await centrifuge.getPresence(address); }; +const publish = (address: LiveChannelAddress, data: unknown, options?: LivePublishOptions) => + centrifuge.publish(address, data, options); + const workObj = { initialize, getConnectionState, @@ -49,6 +52,7 @@ const workObj = { getStream, getQueryData, getPresence, + publish, }; export type RemoteCentrifugeService = typeof workObj; diff --git a/public/app/features/live/centrifuge/serviceWorkerProxy.ts b/public/app/features/live/centrifuge/serviceWorkerProxy.ts index efb87e3cdd5..6f1b8fcc92d 100644 --- a/public/app/features/live/centrifuge/serviceWorkerProxy.ts +++ b/public/app/features/live/centrifuge/serviceWorkerProxy.ts @@ -47,4 +47,8 @@ export class CentrifugeServiceWorkerProxy implements CentrifugeSrv { this.centrifugeWorker.getStream(address) as Promise>>> ); }; + + publish: CentrifugeSrv['publish'] = (address, data, options) => { + return this.centrifugeWorker.publish(address, data, options); + }; } diff --git a/public/app/features/live/live.ts b/public/app/features/live/live.ts index e4ecc3699ea..43f665f5276 100644 --- a/public/app/features/live/live.ts +++ b/public/app/features/live/live.ts @@ -93,7 +93,11 @@ export class GrafanaLiveService implements GrafanaLiveSrv { * * @alpha -- experimental */ - publish: GrafanaLiveSrv['publish'] = async (address, data) => { + publish: GrafanaLiveSrv['publish'] = async (address, data, options) => { + if (options?.useSocket) { + return this.deps.centrifugeSrv.publish(address, data); + } + return this.deps.backendSrv.post(`api/live/publish`, { channel: toLiveChannelId(address), // orgId is from user data,