From 93a01eb1d91e68581d8e2e14d8afad2122bc61b3 Mon Sep 17 00:00:00 2001 From: Ryan McKinley Date: Wed, 23 Sep 2020 08:02:01 -0700 Subject: [PATCH] Live: improved channel state (#27672) --- packages/grafana-data/src/types/datasource.ts | 10 + packages/grafana-data/src/types/index.ts | 1 + packages/grafana-data/src/types/live.ts | 172 ++++++++++++++ packages/grafana-data/src/types/plugin.ts | 17 ++ packages/grafana-runtime/src/services/live.ts | 46 ++-- pkg/api/http_server.go | 4 +- public/app/features/admin/LiveAdmin.tsx | 189 +++++++++++----- public/app/features/admin/LivePanel.tsx | 103 +++++++-- public/app/features/live/channel.ts | 209 ++++++++++++++++++ public/app/features/live/features.ts | 47 ++++ public/app/features/live/live.ts | 186 ++++++++-------- public/app/features/live/scopes.ts | 146 ++++++++++++ public/app/features/plugins/PluginPage.tsx | 2 +- 13 files changed, 914 insertions(+), 218 deletions(-) create mode 100644 packages/grafana-data/src/types/live.ts create mode 100644 public/app/features/live/channel.ts create mode 100644 public/app/features/live/features.ts create mode 100644 public/app/features/live/scopes.ts diff --git a/packages/grafana-data/src/types/datasource.ts b/packages/grafana-data/src/types/datasource.ts index 16d9b6b07dc..2d3c4b482b8 100644 --- a/packages/grafana-data/src/types/datasource.ts +++ b/packages/grafana-data/src/types/datasource.ts @@ -9,6 +9,7 @@ import { DataFrame, DataFrameDTO } from './dataFrame'; import { RawTimeRange, TimeRange } from './time'; import { ScopedVars } from './ScopedVars'; import { CoreApp } from './app'; +import { LiveChannelSupport } from './live'; export interface DataSourcePluginOptionsEditorProps { options: DataSourceSettings; @@ -284,6 +285,15 @@ export abstract class DataSourceApi< * @deprecated -- prefer using {@link AnnotationSupport} */ annotationQuery?(options: AnnotationQueryRequest): Promise; + + /** + * Define live streaming behavior within this datasource settings + * + * Note: `plugin.json` must also define `live: true` + * + * @experimental + */ + channelSupport?: LiveChannelSupport; } export interface MetadataInspectorProps< diff --git a/packages/grafana-data/src/types/index.ts b/packages/grafana-data/src/types/index.ts index 37bf4af4109..75699c2d485 100644 --- a/packages/grafana-data/src/types/index.ts +++ b/packages/grafana-data/src/types/index.ts @@ -27,6 +27,7 @@ export * from './orgs'; export * from './flot'; export * from './trace'; export * from './explore'; +export * from './live'; import * as AppEvents from './appEvents'; import { AppEvent } from './appEvents'; diff --git a/packages/grafana-data/src/types/live.ts b/packages/grafana-data/src/types/live.ts new file mode 100644 index 00000000000..aef0cb8b95b --- /dev/null +++ b/packages/grafana-data/src/types/live.ts @@ -0,0 +1,172 @@ +import { SelectableValue } from './select'; +import { Observable } from 'rxjs'; + +/** + * The channel id is defined as: + * + * ${scope}/${namespace}/${path} + * + * The scope drives how the namespace is used and controlled + */ +export enum LiveChannelScope { + DataSource = 'ds', // namespace = data source ID + Plugin = 'plugin', // namespace = plugin name (singleton works for apps too) + Grafana = 'grafana', // namespace = feature +} + +/** + * @experimental + */ +export interface LiveChannelConfig { + /** + * The path definition. either static, or it may contain variables identifed with {varname} + */ + path: string; + + /** + * An optional description for the channel + */ + description?: string; + + /** + * When variables exist, this list will identify each one + */ + variables?: Array>; + + /** + * The channel keeps track of who else is connected to the same channel + */ + hasPresense?: boolean; + + /** + * This method will be defined if it is possible to publish in this channel. + * The function will return true/false if the current user can publish + */ + canPublish?: () => boolean; + + /** convert the raw stream message into a message that should be broadcast */ + processMessage?: (msg: any) => TMessage; +} + +export enum LiveChannelConnectionState { + /** The connection is not yet established */ + Pending = 'pending', + /** Connected to the channel */ + Connected = 'connected', + /** Disconnected from the channel. The channel will reconnect when possible */ + Disconnected = 'disconnected', + /** Was at some point connected, and will not try to reconnect */ + Shutdown = 'shutdown', + /** Channel configuraiton was invalid and will not connect */ + Invalid = 'invalid', +} + +/** + * @experimental + */ +export interface LiveChannelStatus { + /** + * {scope}/{namespace}/{path} + */ + id: string; + + /** + * unix millies timestamp for the last status change + */ + timestamp: number; + + /** + * flag if the channel is activly connected to the channel. + * This may be false while the connections get established or if the network is lost + * As long as the `shutdown` flag is not set, the connection will try to reestablish + */ + state: LiveChannelConnectionState; + + /** + * The last error. + * + * This will remain in the status until a new message is succesfully recieved from the channel + */ + error?: any; +} + +/** + * @experimental + */ +export interface LiveChannelJoinLeave { + user: any; +} + +/** + * @experimental + */ +export interface LiveChannelPresense { + users: any; +} + +export interface LiveChannelMessage { + type: 'status' | 'message' | 'join' | 'leave'; + message: TMessage | LiveChannelStatus | LiveChannelJoinLeave; +} + +/** + * @experimental + */ +export interface LiveChannel { + /** The fully qualified channel id: ${scope}/${namespace}/${path} */ + id: string; + + /** The scope for this channel */ + scope: LiveChannelScope; + + /** datasourceId/plugin name/feature depending on scope */ + namespace: string; + + /** additional qualifier */ + path: string; + + /** Unix timestamp for when the channel connected */ + opened: number; + + /** Static definition of the channel definition. This may describe the channel usage */ + config?: LiveChannelConfig; + + /** + * Watch all events in this channel + */ + getStream: () => Observable>; + + /** + * For channels that support presense, this will request the current state from the server. + * + * Join and leave messages will be sent to the open stream + */ + getPresense?: () => Promise; + + /** + * Write a message into the channel + * + * NOTE: This feature is supported by a limited set of channels + */ + publish?: (msg: TPublish) => Promise; + + /** + * This will close and terminate this channel + */ + disconnect: () => void; +} + +/** + * @experimental + */ +export interface LiveChannelSupport { + /** + * Get the channel handler for the path, or throw an error if invalid + */ + getChannelConfig(path: string): LiveChannelConfig | undefined; + + /** + * Return a list of supported channels + */ + getSupportedPaths(): LiveChannelConfig[]; +} diff --git a/packages/grafana-data/src/types/plugin.ts b/packages/grafana-data/src/types/plugin.ts index 745b971b0ca..0aa96f68ed8 100644 --- a/packages/grafana-data/src/types/plugin.ts +++ b/packages/grafana-data/src/types/plugin.ts @@ -1,5 +1,6 @@ import { ComponentClass } from 'react'; import { KeyValue } from './data'; +import { LiveChannelSupport } from './live'; export enum PluginState { alpha = 'alpha', // Only included it `enable_alpha` is true @@ -47,6 +48,7 @@ export interface PluginMeta { latestVersion?: string; pinned?: boolean; signature?: PluginSignatureStatus; + live?: boolean; } interface PluginDependencyInfo { @@ -139,6 +141,13 @@ export class GrafanaPlugin { // This is set if the plugin system had errors loading the plugin loadError?: boolean; + /** + * Live streaming support + * + * Note: `plugin.json` must also define `live: true` + */ + channelSupport?: LiveChannelSupport; + // Config control (app/datasource) angularConfigCtrl?: any; @@ -154,6 +163,14 @@ export class GrafanaPlugin { return this; } + /** + * Specify how the plugin should support paths within the live streaming environment + */ + setChannelSupport(support: LiveChannelSupport) { + this.channelSupport = support; + return this; + } + constructor() { this.meta = {} as T; } diff --git a/packages/grafana-runtime/src/services/live.ts b/packages/grafana-runtime/src/services/live.ts index 18f596a2bc8..94f0d3663be 100644 --- a/packages/grafana-runtime/src/services/live.ts +++ b/packages/grafana-runtime/src/services/live.ts @@ -1,25 +1,6 @@ +import { LiveChannel, LiveChannelScope } from '@grafana/data'; import { Observable } from 'rxjs'; -/** - * @experimental - */ -export interface ChannelHandler { - /** - * Process the raw message from the server before broadcasting it - * to all subscribeers on this channel - */ - onPublish(msg: any): T; -} - -// export interface SubscriptionEvents { -// publish?: (ctx: PublicationContext) => void; -// join?: (ctx: JoinLeaveContext) => void; -// leave?: (ctx: JoinLeaveContext) => void; -// subscribe?: (ctx: SubscribeSuccessContext) => void; -// error?: (ctx: SubscribeErrorContext) => void; -// unsubscribe?: (ctx: UnsubscribeContext) => void; -// } - /** * @experimental */ @@ -30,24 +11,23 @@ export interface GrafanaLiveSrv { isConnected(): boolean; /** - * Listen for changes to the connection state + * Listen for changes to the main service */ getConnectionState(): Observable; /** - * Configure a channel with the given setup + * Get a channel. If the scope, namespace, or path is invalid, a shutdown + * channel will be returned with an error state indicated in its status. + * + * This is a singleton instance that stays active until explicitly shutdown. + * Multiple requests for this channel will return the same object until + * the channel is shutdown */ - initChannel(channel: string, handler: ChannelHandler): void; - - /** - * Subscribe to activity on a given channel - */ - getChannelStream(channel: string): Observable; - - /** - * Send data to a channel. This feature is disabled for most channels and will return an error - */ - publish(channel: string, data: any): Promise; + getChannel( + scope: LiveChannelScope, + namespace: string, + path: string + ): LiveChannel; } let singletonInstance: GrafanaLiveSrv; diff --git a/pkg/api/http_server.go b/pkg/api/http_server.go index 19f416831f1..237467b52bc 100644 --- a/pkg/api/http_server.go +++ b/pkg/api/http_server.go @@ -86,8 +86,8 @@ func (hs *HTTPServer) Init() error { hs.Live = node // Spit random walk to example - go live.RunRandomCSV(hs.Live, "random-2s-stream", 2000, 0) - go live.RunRandomCSV(hs.Live, "random-flakey-stream", 400, .6) + go live.RunRandomCSV(hs.Live, "grafana/testdata/random-2s-stream", 2000, 0) + go live.RunRandomCSV(hs.Live, "grafana/testdata/random-flakey-stream", 400, .6) } hs.macaron = hs.newMacaron() diff --git a/public/app/features/admin/LiveAdmin.tsx b/public/app/features/admin/LiveAdmin.tsx index dc6e1486df9..cb197239501 100644 --- a/public/app/features/admin/LiveAdmin.tsx +++ b/public/app/features/admin/LiveAdmin.tsx @@ -1,85 +1,137 @@ import React, { PureComponent } from 'react'; import { hot } from 'react-hot-loader'; import { connect } from 'react-redux'; +import { css } from 'emotion'; import { StoreState } from 'app/types'; import { getNavModel } from 'app/core/selectors/navModel'; import Page from 'app/core/components/Page/Page'; -import { NavModel, SelectableValue, FeatureState } from '@grafana/data'; +import { + NavModel, + SelectableValue, + FeatureState, + LiveChannelScope, + LiveChannelConfig, + LiveChannelSupport, +} from '@grafana/data'; import { LivePanel } from './LivePanel'; -import { Select, Input, Button, FeatureInfoBox, Container } from '@grafana/ui'; -import { getGrafanaLiveSrv } from '@grafana/runtime'; +import { Select, FeatureInfoBox, Container } from '@grafana/ui'; +import { getGrafanaLiveCentrifugeSrv } from '../live/live'; interface Props { navModel: NavModel; } +const scopes: Array> = [ + { label: 'Grafana', value: LiveChannelScope.Grafana, description: 'Core grafana live features' }, + { label: 'Data Sources', value: LiveChannelScope.DataSource, description: 'Data sources with live support' }, + { label: 'Plugins', value: LiveChannelScope.Plugin, description: 'Plugins with live support' }, +]; + interface State { - channel: string; - text: string; + scope: LiveChannelScope; + namespace?: string; + path?: string; + + namespaces: Array>; + paths: Array>; + support?: LiveChannelSupport; + config?: LiveChannelConfig; } export class LiveAdmin extends PureComponent { state: State = { - channel: 'random-2s-stream', - text: '', // publish text to a channel + scope: LiveChannelScope.Grafana, + namespace: 'testdata', + path: 'random-2s-stream', + namespaces: [], + paths: [], }; + // onTextChanged: ((event: FormEvent) => void) | undefined; + // onPublish: ((event: MouseEvent) => void) | undefined; - onChannelChanged = (v: SelectableValue) => { + async componentDidMount() { + const { scope, namespace, path } = this.state; + const srv = getGrafanaLiveCentrifugeSrv(); + const namespaces = await srv.scopes[scope].listNamespaces(); + const support = namespace ? await srv.scopes[scope].getChannelSupport(namespace) : undefined; + const paths = support ? await support.getSupportedPaths() : undefined; + const config = support && path ? await support.getChannelConfig(path) : undefined; + + this.setState({ + namespaces, + support, + paths: paths + ? paths.map(p => ({ + label: p.path, + value: p.path, + description: p.description, + })) + : [], + config, + }); + } + + onScopeChanged = async (v: SelectableValue) => { if (v.value) { + const srv = getGrafanaLiveCentrifugeSrv(); + this.setState({ - channel: v.value, + scope: v.value, + namespace: undefined, + path: undefined, + namespaces: await srv.scopes[v.value!].listNamespaces(), + paths: [], + support: undefined, + config: undefined, }); } }; - onTextChanged = (event: React.ChangeEvent) => { - this.setState({ text: event.target.value }); - }; + onNamespaceChanged = async (v: SelectableValue) => { + if (v.value) { + const namespace = v.value; + const srv = getGrafanaLiveCentrifugeSrv(); + const support = await srv.scopes[this.state.scope].getChannelSupport(namespace); - onPublish = () => { - const { text, channel } = this.state; - if (text) { - const msg = { - line: text, - }; + this.setState({ + namespace: v.value, + paths: support!.getSupportedPaths().map(p => ({ + label: p.path, + value: p.path, + description: p.description, + })), + path: undefined, + config: undefined, + }); + } + }; - const srv = getGrafanaLiveSrv(); - srv.publish(channel, msg).then(v => { - console.log('PUBLISHED', text, v); + onPathChanged = async (v: SelectableValue) => { + if (v.value) { + const path = v.value; + const srv = getGrafanaLiveCentrifugeSrv(); + const support = await srv.scopes[this.state.scope].getChannelSupport(this.state.namespace!); + if (!support) { + this.setState({ + namespace: undefined, + paths: [], + config: undefined, + support, + }); + return; + } + + this.setState({ + path, + support, + config: support.getChannelConfig(path), }); } - this.setState({ text: '' }); }; render() { const { navModel } = this.props; - const { channel, text } = this.state; - - const channels: Array> = [ - { - label: 'random-2s-stream', - value: 'random-2s-stream', - description: 'Random stream that updates every 2s', - }, - { - label: 'random-flakey-stream', - value: 'random-flakey-stream', - description: 'Random stream with intermittent updates', - }, - { - label: 'example-chat', - value: 'example-chat', - description: 'A channel that expects chat messages', - }, - ]; - let current = channels.find(f => f.value === channel); - if (!current) { - current = { - label: channel, - value: channel, - }; - channels.push(current); - } + const { scope, namespace, namespaces, path, paths, config } = this.state; return ( @@ -99,19 +151,36 @@ export class LiveAdmin extends PureComponent {
-

Channels

- s.value === scope)} onChange={this.onScopeChanged} /> + +
+
Namespace
+ s.value === path) || ''} onChange={this.onPathChanged} /> +
+

-

Write to channel

- - + {scope && namespace && path && } ); diff --git a/public/app/features/admin/LivePanel.tsx b/public/app/features/admin/LivePanel.tsx index ce544a87b71..fbfd423bcfd 100644 --- a/public/app/features/admin/LivePanel.tsx +++ b/public/app/features/admin/LivePanel.tsx @@ -1,52 +1,71 @@ import React, { PureComponent } from 'react'; import { Unsubscribable, PartialObserver } from 'rxjs'; import { getGrafanaLiveSrv } from '@grafana/runtime'; +import { + AppEvents, + LiveChannel, + LiveChannelConfig, + LiveChannelConnectionState, + LiveChannelMessage, + LiveChannelScope, + LiveChannelStatus, +} from '@grafana/data'; +import { Input, Button } from '@grafana/ui'; +import { appEvents } from 'app/core/core'; interface Props { - channel: string; + scope: LiveChannelScope; + namespace: string; + path: string; + config?: LiveChannelConfig; } interface State { - connected: boolean; + channel?: LiveChannel; + status: LiveChannelStatus; count: number; lastTime: number; lastBody: string; + text: string; // for publish! } export class LivePanel extends PureComponent { state: State = { - connected: false, + status: { id: '?', state: LiveChannelConnectionState.Pending, timestamp: Date.now() }, count: 0, lastTime: 0, lastBody: '', + text: '', }; subscription?: Unsubscribable; - observer: PartialObserver = { - next: (msg: any) => { - this.setState({ - count: this.state.count + 1, - lastTime: Date.now(), - lastBody: JSON.stringify(msg), - }); + streamObserver: PartialObserver = { + next: (msg: LiveChannelMessage) => { + if (msg.type === 'status') { + this.setState({ status: msg.message as LiveChannelStatus }); + } else { + this.setState({ + count: this.state.count + 1, + lastTime: Date.now(), + lastBody: JSON.stringify(msg), + }); + } }, }; startSubscription = () => { - if (this.subscription) { - this.subscription.unsubscribe(); - this.subscription = undefined; + const { scope, namespace, path } = this.props; + const channel = getGrafanaLiveSrv().getChannel(scope, namespace, path); + if (this.state.channel === channel) { + return; // no change! } - const srv = getGrafanaLiveSrv(); - if (srv.isConnected()) { - const stream = srv.getChannelStream(this.props.channel); - this.subscription = stream.subscribe(this.observer); - this.setState({ connected: true, count: 0, lastTime: 0, lastBody: '' }); - return; + if (this.subscription) { + this.subscription.unsubscribe(); } - console.log('Not yet connected... try again...'); - setTimeout(this.startSubscription, 200); + + this.subscription = channel.getStream().subscribe(this.streamObserver); + this.setState({ channel }); }; componentDidMount = () => { @@ -56,21 +75,47 @@ export class LivePanel extends PureComponent { componentWillUnmount() { if (this.subscription) { this.subscription.unsubscribe(); - this.subscription = undefined; } } componentDidUpdate(oldProps: Props) { - if (oldProps.channel !== this.props.channel) { + if (oldProps.config !== this.props.config) { this.startSubscription(); } } + onTextChanged = (event: React.ChangeEvent) => { + this.setState({ text: event.target.value }); + }; + + onPublish = () => { + const { text, channel } = this.state; + if (text && channel) { + const msg = { + line: text, + }; + + channel.publish!(msg) + .then(v => { + console.log('PUBLISHED', text, v); + }) + .catch(err => { + appEvents.emit(AppEvents.alertError, ['Publish error', `${err}`]); + }); + } + this.setState({ text: '' }); + }; + render() { - const { lastBody, lastTime, count } = this.state; + const { lastBody, lastTime, count, status, text } = this.state; + const { config } = this.props; + const showPublish = config && config.canPublish && config.canPublish(); return (
+
Status: {config ? '' : '(no config)'}
+ 
{JSON.stringify(status)}
+
Count: {count}
{lastTime > 0 && ( <> @@ -82,6 +127,16 @@ export class LivePanel extends PureComponent { )} )} + + {showPublish && ( +
+

Write to channel

+ + +
+ )}
); } diff --git a/public/app/features/live/channel.ts b/public/app/features/live/channel.ts new file mode 100644 index 00000000000..dd3870784d4 --- /dev/null +++ b/public/app/features/live/channel.ts @@ -0,0 +1,209 @@ +import { + LiveChannelConfig, + LiveChannel, + LiveChannelScope, + LiveChannelStatus, + LiveChannelPresense, + LiveChannelJoinLeave, + LiveChannelMessage, + LiveChannelConnectionState, +} from '@grafana/data'; +import Centrifuge, { + JoinLeaveContext, + PublicationContext, + SubscribeErrorContext, + SubscribeSuccessContext, + SubscriptionEvents, + UnsubscribeContext, +} from 'centrifuge/dist/centrifuge.protobuf'; +import { Subject, of, merge } from 'rxjs'; + +/** + * Internal class that maps Centrifuge support to GrafanaLive + */ +export class CentrifugeLiveChannel implements LiveChannel { + readonly currentStatus: LiveChannelStatus; + + readonly opened = Date.now(); + readonly id: string; + readonly scope: LiveChannelScope; + readonly namespace: string; + readonly path: string; + + readonly stream = new Subject>(); + + // When presense is enabled (rarely), this will be initalized + private presense?: Subject; + + /** Static definition of the channel definition. This may describe the channel usage */ + config?: LiveChannelConfig; + subscription?: Centrifuge.Subscription; + shutdownCallback?: () => void; + + constructor(id: string, scope: LiveChannelScope, namespace: string, path: string) { + this.id = id; + this.scope = scope; + this.namespace = namespace; + this.path = path; + this.currentStatus = { + id, + timestamp: this.opened, + state: LiveChannelConnectionState.Pending, + }; + } + + // This should only be called when centrifuge is connected + initalize(config: LiveChannelConfig): SubscriptionEvents { + if (this.config) { + throw new Error('Channel already initalized: ' + this.id); + } + this.config = config; + const prepare = config.processMessage ? config.processMessage : (v: any) => v; + + const events: SubscriptionEvents = { + // This means a message was recieved from the server + publish: (ctx: PublicationContext) => { + this.stream.next(prepare(ctx.data)); + + // Clear any error messages + if (this.currentStatus.error) { + this.currentStatus.timestamp = Date.now(); + delete this.currentStatus.error; + this.sendStatus(); + } + }, + error: (ctx: SubscribeErrorContext) => { + this.currentStatus.timestamp = Date.now(); + this.currentStatus.error = ctx.error; + this.sendStatus(); + }, + subscribe: (ctx: SubscribeSuccessContext) => { + this.currentStatus.timestamp = Date.now(); + this.currentStatus.state = LiveChannelConnectionState.Connected; + this.sendStatus(); + }, + unsubscribe: (ctx: UnsubscribeContext) => { + this.currentStatus.timestamp = Date.now(); + this.currentStatus.state = LiveChannelConnectionState.Disconnected; + this.sendStatus(); + }, + }; + + if (config.hasPresense) { + events.join = (ctx: JoinLeaveContext) => { + const message: LiveChannelJoinLeave = { + user: ctx.info.user, + }; + this.stream.next({ + type: 'join', + message, + }); + }; + events.leave = (ctx: JoinLeaveContext) => { + const message: LiveChannelJoinLeave = { + user: ctx.info.user, + }; + this.stream.next({ + type: 'leave', + message, + }); + }; + + this.getPresense = () => { + return this.subscription!.presence().then(v => { + return { + users: Object.keys(v.presence), + }; + }); + }; + } + return events; + } + + private sendStatus() { + this.stream.next({ type: 'status', message: { ...this.currentStatus } }); + } + + /** + * Get the stream of events and + */ + getStream() { + const status: LiveChannelMessage = { type: 'status', message: { ...this.currentStatus } }; + return merge(of(status), this.stream.asObservable()); + } + + /** + * This is configured by the server when the config supports presense + */ + getPresense?: () => Promise; + + /** + * This is configured by the server when config supports writing + */ + publish?: (msg: TPublish) => Promise; + + /** + * This will close and terminate all streams for this channel + */ + disconnect() { + this.currentStatus.state = LiveChannelConnectionState.Shutdown; + this.currentStatus.timestamp = Date.now(); + + if (this.subscription) { + this.subscription.unsubscribe(); + this.subscription.removeAllListeners(); // they keep all listeners attached after unsubscribe + this.subscription = undefined; + } + + this.stream.complete(); + + if (this.presense) { + this.presense.complete(); + } + + this.stream.next({ type: 'status', message: { ...this.currentStatus } }); + this.stream.complete(); + + if (this.shutdownCallback) { + this.shutdownCallback(); + } + } + + shutdownWithError(err: string) { + this.currentStatus.error = err; + this.disconnect(); + } +} + +export function getErrorChannel( + msg: string, + id: string, + scope: LiveChannelScope, + namespace: string, + path: string +): LiveChannel { + const errorStatus: LiveChannelStatus = { + id, + timestamp: Date.now(), + state: LiveChannelConnectionState.Invalid, + error: msg, + }; + + return { + id, + opened: Date.now(), + scope, + namespace, + path, + + // return an error + getStream: () => + of({ + type: 'status', + message: errorStatus, + }), + + // already disconnected + disconnect: () => {}, + }; +} diff --git a/public/app/features/live/features.ts b/public/app/features/live/features.ts new file mode 100644 index 00000000000..6d7eb21c80a --- /dev/null +++ b/public/app/features/live/features.ts @@ -0,0 +1,47 @@ +import { LiveChannelConfig } from '@grafana/data'; +import { grafanaLiveCoreFeatures } from './scopes'; + +export function registerLiveFeatures() { + const channels = [ + { + path: 'random-2s-stream', + description: 'Random stream with points every 2s', + }, + { + path: 'random-flakey-stream', + description: 'Random stream with flakey data points', + }, + ]; + + grafanaLiveCoreFeatures.register( + 'testdata', + { + getChannelConfig: (path: string) => { + return channels.find(c => c.path === path); + }, + getSupportedPaths: () => channels, + }, + 'Test data generations' + ); + + const chatConfig: LiveChannelConfig = { + path: 'chat', + description: 'Broadcast text messages to a channel', + canPublish: () => true, + hasPresense: true, + }; + + grafanaLiveCoreFeatures.register( + 'experimental', + { + getChannelConfig: (path: string) => { + if ('chat' === path) { + return chatConfig; + } + throw new Error('invalid path: ' + path); + }, + getSupportedPaths: () => [chatConfig], + }, + 'Experimental features' + ); +} diff --git a/public/app/features/live/live.ts b/public/app/features/live/live.ts index 8b2d580dd14..c0a4985b355 100644 --- a/public/app/features/live/live.ts +++ b/public/app/features/live/live.ts @@ -1,26 +1,24 @@ -import Centrifuge, { - PublicationContext, - SubscriptionEvents, - SubscribeSuccessContext, - UnsubscribeContext, - JoinLeaveContext, - SubscribeErrorContext, -} from 'centrifuge/dist/centrifuge.protobuf'; +import Centrifuge from 'centrifuge/dist/centrifuge.protobuf'; import SockJS from 'sockjs-client'; -import { GrafanaLiveSrv, setGrafanaLiveSrv, ChannelHandler, config } from '@grafana/runtime'; -import { Observable, Subject, BehaviorSubject } from 'rxjs'; -import { KeyValue } from '@grafana/data'; +import { GrafanaLiveSrv, setGrafanaLiveSrv, getGrafanaLiveSrv, config } from '@grafana/runtime'; +import { BehaviorSubject } from 'rxjs'; +import { LiveChannel, LiveChannelScope } from '@grafana/data'; +import { CentrifugeLiveChannel, getErrorChannel } from './channel'; +import { + GrafanaLiveScope, + grafanaLiveCoreFeatures, + GrafanaLiveDataSourceScope, + GrafanaLivePluginScope, +} from './scopes'; +import { registerLiveFeatures } from './features'; -interface Channel { - subject: Subject; - subscription?: Centrifuge.Subscription; -} +export class CentrifugeSrv implements GrafanaLiveSrv { + readonly open = new Map(); -class CentrifugeSrv implements GrafanaLiveSrv { - centrifuge: Centrifuge; - channels: KeyValue = {}; - connectionState: BehaviorSubject; - standardCallbacks: SubscriptionEvents; + readonly centrifuge: Centrifuge; + readonly connectionState: BehaviorSubject; + readonly connectionBlocker: Promise; + readonly scopes: Record; constructor() { this.centrifuge = new Centrifuge(`${config.appUrl}live/sockjs`, { @@ -29,19 +27,27 @@ class CentrifugeSrv implements GrafanaLiveSrv { }); this.centrifuge.connect(); // do connection this.connectionState = new BehaviorSubject(this.centrifuge.isConnected()); + this.connectionBlocker = new Promise(resolve => { + if (this.centrifuge.isConnected()) { + return resolve(); + } + const connectListener = () => { + resolve(); + this.centrifuge.removeListener('connect', connectListener); + }; + this.centrifuge.addListener('connect', connectListener); + }); + + this.scopes = { + [LiveChannelScope.Grafana]: grafanaLiveCoreFeatures, + [LiveChannelScope.DataSource]: new GrafanaLiveDataSourceScope(), + [LiveChannelScope.Plugin]: new GrafanaLivePluginScope(), + }; // Register global listeners this.centrifuge.on('connect', this.onConnect); this.centrifuge.on('disconnect', this.onDisconnect); this.centrifuge.on('publish', this.onServerSideMessage); - - this.standardCallbacks = { - subscribe: this.onSubscribe, - unsubscribe: this.onUnsubscribe, - join: this.onJoin, - leave: this.onLeave, - error: this.onError, - }; } //---------------------------------------------------------- @@ -62,38 +68,61 @@ class CentrifugeSrv implements GrafanaLiveSrv { console.log('Publication from server-side channel', context); }; - //---------------------------------------------------------- - // Channel functions - //---------------------------------------------------------- + /** + * Get a channel. If the scope, namespace, or path is invalid, a shutdown + * channel will be returned with an error state indicated in its status + */ + getChannel( + scopeId: LiveChannelScope, + namespace: string, + path: string + ): LiveChannel { + const id = `${scopeId}/${namespace}/${path}`; + let channel = this.open.get(id); + if (channel != null) { + return channel; + } - // export interface SubscriptionEvents { - // publish?: (ctx: PublicationContext) => void; - // join?: (ctx: JoinLeaveContext) => void; - // leave?: (ctx: JoinLeaveContex) => void; - // subscribe?: (ctx: SubscribeSuccessContext) => void; - // error?: (ctx: SubscribeErrorContext) => void; - // unsubscribe?: (ctx: UnsubscribeContext) => void; - // } + const scope = this.scopes[scopeId]; + if (!scope) { + return getErrorChannel('invalid scope', id, scopeId, namespace, path); + } - onSubscribe = (context: SubscribeSuccessContext) => { - console.log('onSubscribe', context); - }; + channel = new CentrifugeLiveChannel(id, scopeId, namespace, path); + channel.shutdownCallback = () => { + this.open.delete(id); // remove it from the list of open channels + }; + this.open.set(id, channel); - onUnsubscribe = (context: UnsubscribeContext) => { - console.log('onUnsubscribe', context); - }; + // Initalize the channel in the bacground + this.initChannel(scope, channel).catch(err => { + channel?.shutdownWithError(err); + this.open.delete(id); + }); - onJoin = (context: JoinLeaveContext) => { - console.log('onJoin', context); - }; + // return the not-yet initalized channel + return channel; + } - onLeave = (context: JoinLeaveContext) => { - console.log('onLeave', context); - }; - - onError = (context: SubscribeErrorContext) => { - console.log('onError', context); - }; + private async initChannel(scope: GrafanaLiveScope, channel: CentrifugeLiveChannel): Promise { + const support = await scope.getChannelSupport(channel.namespace); + if (!support) { + throw new Error(channel.namespace + 'does not support streaming'); + } + const config = support.getChannelConfig(channel.path); + if (!config) { + throw new Error('unknown path: ' + channel.path); + } + const events = channel.initalize(config); + if (!this.centrifuge.isConnected()) { + await this.connectionBlocker; + } + if (config.canPublish && config.canPublish()) { + channel.publish = (data: any) => this.centrifuge.publish(channel.id, data); + } + channel.subscription = this.centrifuge.subscribe(channel.id, events); + return; + } //---------------------------------------------------------- // Exported functions @@ -112,52 +141,13 @@ class CentrifugeSrv implements GrafanaLiveSrv { getConnectionState() { return this.connectionState.asObservable(); } - - initChannel(path: string, handler: ChannelHandler) { - if (this.channels[path]) { - console.log('Already connected to:', path); - return; - } - const c: Channel = { - subject: new Subject(), - }; - this.channels[path] = c; - - console.log('initChannel', this.centrifuge.isConnected(), path, handler); - const callbacks: SubscriptionEvents = { - ...this.standardCallbacks, - publish: (ctx: PublicationContext) => { - // console.log('GOT', JSON.stringify(ctx.data), ctx); - const v = handler.onPublish(ctx.data); - c.subject.next(v); - }, - }; - c.subscription = this.centrifuge.subscribe(path, callbacks); - } - - getChannelStream(path: string): Observable { - let c = this.channels[path]; - if (!c) { - this.initChannel(path, noopChannelHandler); - c = this.channels[path]; - } - return c!.subject.asObservable(); - } - - /** - * Send data to a channel. This feature is disabled for most channels and will return an error - */ - publish(channel: string, data: any): Promise { - return this.centrifuge.publish(channel, data); - } } -const noopChannelHandler: ChannelHandler = { - onPublish: (v: any) => { - return v; // Just pass the object along - }, -}; +export function getGrafanaLiveCentrifugeSrv() { + return getGrafanaLiveSrv() as CentrifugeSrv; +} export function initGrafanaLive() { setGrafanaLiveSrv(new CentrifugeSrv()); + registerLiveFeatures(); } diff --git a/public/app/features/live/scopes.ts b/public/app/features/live/scopes.ts new file mode 100644 index 00000000000..30f8a12b29b --- /dev/null +++ b/public/app/features/live/scopes.ts @@ -0,0 +1,146 @@ +import { LiveChannelScope, LiveChannelSupport, SelectableValue } from '@grafana/data'; +import { getDataSourceSrv } from '@grafana/runtime'; +import { config } from 'app/core/config'; +import { loadPlugin } from '../plugins/PluginPage'; + +export abstract class GrafanaLiveScope { + constructor(protected scope: LiveChannelScope) {} + + /** + * Load the real namespaces + */ + abstract async getChannelSupport(namespace: string): Promise; + + /** + * List the possible values within this scope + */ + abstract async listNamespaces(): Promise>>; +} + +class GrafanaLiveCoreScope extends GrafanaLiveScope { + readonly features = new Map(); + readonly namespaces: Array> = []; + + constructor() { + super(LiveChannelScope.Grafana); + } + + register(feature: string, support: LiveChannelSupport, description: string): GrafanaLiveCoreScope { + this.features.set(feature, support); + this.namespaces.push({ + value: feature, + label: feature, + description, + }); + return this; + } + + /** + * Load the real namespaces + */ + async getChannelSupport(namespace: string) { + const v = this.features.get(namespace); + if (v) { + return Promise.resolve(v); + } + throw new Error('unknown feature: ' + namespace); + } + + /** + * List the possible values within this scope + */ + listNamespaces() { + return Promise.resolve(this.namespaces); + } +} +export const grafanaLiveCoreFeatures = new GrafanaLiveCoreScope(); + +export class GrafanaLiveDataSourceScope extends GrafanaLiveScope { + names?: Array>; + + constructor() { + super(LiveChannelScope.DataSource); + } + + /** + * Load the real namespaces + */ + async getChannelSupport(namespace: string) { + const ds = await getDataSourceSrv().get(namespace); + return ds.channelSupport; + } + + /** + * List the possible values within this scope + */ + async listNamespaces() { + if (this.names) { + return Promise.resolve(this.names); + } + const names: Array> = []; + for (const [key, ds] of Object.entries(config.datasources)) { + if (ds.meta.live) { + try { + const s = this.getChannelSupport(key); // ds.name or ID? + if (s) { + names.push({ + label: ds.name, + value: ds.type, + description: ds.type, + }); + } + } catch (err) { + err.isHandled = true; + } + } + } + return (this.names = names); + } +} + +export class GrafanaLivePluginScope extends GrafanaLiveScope { + names?: Array>; + + constructor() { + super(LiveChannelScope.Plugin); + } + + /** + * Load the real namespaces + */ + async getChannelSupport(namespace: string) { + const plugin = await loadPlugin(namespace); + if (!plugin.channelSupport) { + throw new Error('Unknown plugin: ' + namespace); + } + return plugin.channelSupport; + } + + /** + * List the possible values within this scope + */ + async listNamespaces() { + if (this.names) { + return Promise.resolve(this.names); + } + const names: Array> = []; + // TODO add list to config + for (const [key, panel] of Object.entries(config.panels)) { + if (panel.live) { + try { + const s = this.getChannelSupport(key); // ds.name or ID? + if (s) { + names.push({ + label: panel.name, + value: key, + description: panel.info?.description, + }); + } + } catch (err) { + err.isHandled = true; + } + } + } + return (this.names = names); + } +} diff --git a/public/app/features/plugins/PluginPage.tsx b/public/app/features/plugins/PluginPage.tsx index 6fec9edf38a..372de885a36 100644 --- a/public/app/features/plugins/PluginPage.tsx +++ b/public/app/features/plugins/PluginPage.tsx @@ -42,7 +42,7 @@ export function getLoadingNav(): NavModel { }; } -function loadPlugin(pluginId: string): Promise { +export function loadPlugin(pluginId: string): Promise { return getPluginSettings(pluginId).then(info => { if (info.type === PluginType.app) { return importAppPlugin(info);