From c780854a18100fa9df64dd8b9609452e04f3f72c Mon Sep 17 00:00:00 2001 From: Ryan McKinley Date: Mon, 15 Nov 2021 07:54:14 -0800 Subject: [PATCH] Live: remove hardcoded frontend channel support (#41584) --- packages/grafana-data/src/types/datasource.ts | 10 - packages/grafana-data/src/types/live.ts | 35 ---- packages/grafana-data/src/types/plugin.ts | 14 +- packages/grafana-runtime/src/services/live.ts | 11 -- .../app/features/live/centrifuge/channel.ts | 29 +-- .../app/features/live/centrifuge/service.ts | 30 +-- .../live/centrifuge/service.worker.ts | 14 +- .../live/centrifuge/serviceWorkerProxy.ts | 14 +- .../app/features/live/channel-config/index.ts | 36 ---- .../app/features/live/channel-config/scope.ts | 182 ------------------ .../app/features/live/channel-config/types.ts | 18 -- .../live/dashboard/dashboardWatcher.ts | 14 -- public/app/features/live/features.ts | 34 ---- public/app/features/live/index.ts | 9 - public/app/features/live/live.ts | 74 +------ .../live/measurements/measurementsSupport.ts | 17 -- .../plugins/datasource/cloudwatch/module.tsx | 4 +- .../plugins/datasource/testdata/module.tsx | 2 - .../plugins/panel/live/LiveChannelEditor.tsx | 27 +-- 19 files changed, 54 insertions(+), 520 deletions(-) delete mode 100644 public/app/features/live/channel-config/index.ts delete mode 100644 public/app/features/live/channel-config/scope.ts delete mode 100644 public/app/features/live/channel-config/types.ts delete mode 100644 public/app/features/live/features.ts delete mode 100644 public/app/features/live/measurements/measurementsSupport.ts diff --git a/packages/grafana-data/src/types/datasource.ts b/packages/grafana-data/src/types/datasource.ts index ec5597e4401..c65a07f1674 100644 --- a/packages/grafana-data/src/types/datasource.ts +++ b/packages/grafana-data/src/types/datasource.ts @@ -9,7 +9,6 @@ import { DataFrame, DataFrameDTO } from './dataFrame'; import { RawTimeRange, TimeRange } from './time'; import { ScopedVars } from './ScopedVars'; import { CoreApp } from './app'; -import { LiveChannelSupport } from './live'; import { CustomVariableSupport, DataSourceVariableSupport, StandardVariableSupport } from './variables'; import { makeClassES5Compatible } from '../utils/makeClassES5Compatible'; import { DataQuery } from './query'; @@ -343,15 +342,6 @@ abstract class DataSourceApi< */ annotationQuery?(options: AnnotationQueryRequest): Promise; - /** - * Define live streaming behavior within this datasource settings - * - * Note: `plugin.json` must also define `live: true` - * - * @alpha -- experimental - */ - channelSupport?: LiveChannelSupport; - /** * Defines new variable support * @alpha -- experimental diff --git a/packages/grafana-data/src/types/live.ts b/packages/grafana-data/src/types/live.ts index 3c924d554ca..0319b0b2f9e 100644 --- a/packages/grafana-data/src/types/live.ts +++ b/packages/grafana-data/src/types/live.ts @@ -25,31 +25,6 @@ export enum LiveChannelType { JSON = 'json', // arbitray json message } -/** - * @alpha -- experimental - */ -export interface LiveChannelConfig { - /** - * An optional description for the channel - */ - description?: string; - - /** - * What kind of data do you expect - */ - type?: LiveChannelType; - - /** - * The channel keeps track of who else is connected to the same channel - */ - hasPresence?: boolean; - - /** - * Allow users to write to the connection - */ - canPublish?: boolean; -} - export enum LiveChannelConnectionState { /** The connection is not yet established */ Pending = 'pending', @@ -206,13 +181,3 @@ export function toLiveChannelId(addr: LiveChannelAddress): string { } return id + '/' + addr.path; } - -/** - * @alpha -- experimental - */ -export interface LiveChannelSupport { - /** - * Get the channel handler for the path, or throw an error if invalid - */ - getChannelConfig(path: string): LiveChannelConfig | undefined; -} diff --git a/packages/grafana-data/src/types/plugin.ts b/packages/grafana-data/src/types/plugin.ts index 94d037cea5e..e663be6a0d8 100644 --- a/packages/grafana-data/src/types/plugin.ts +++ b/packages/grafana-data/src/types/plugin.ts @@ -1,6 +1,5 @@ import { ComponentClass } from 'react'; import { KeyValue } from './data'; -import { LiveChannelSupport } from './live'; /** Describes plugins life cycle status */ export enum PluginState { @@ -170,13 +169,6 @@ 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; @@ -193,10 +185,10 @@ export class GrafanaPlugin { } /** - * Specify how the plugin should support paths within the live streaming environment + * @deprecated -- this is no longer necessary and will be removed */ - setChannelSupport(support: LiveChannelSupport) { - this.channelSupport = support; + setChannelSupport(support: any) { + console.warn('[deprecation] plugin is using ignored option: setChannelSupport', this.meta); return this; } diff --git a/packages/grafana-runtime/src/services/live.ts b/packages/grafana-runtime/src/services/live.ts index ce50bd71ecb..199eb3d17af 100644 --- a/packages/grafana-runtime/src/services/live.ts +++ b/packages/grafana-runtime/src/services/live.ts @@ -2,7 +2,6 @@ import { DataFrame, DataQueryResponse, LiveChannelAddress, - LiveChannelConfig, LiveChannelEvent, LiveChannelPresenceStatus, StreamingFrameOptions, @@ -36,16 +35,6 @@ export interface GrafanaLiveSrv { */ getConnectionState(): Observable; - /** - * 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 - */ - getChannelInfo(address: LiveChannelAddress): Promise; - /** * Watch for messages in a channel */ diff --git a/public/app/features/live/centrifuge/channel.ts b/public/app/features/live/centrifuge/channel.ts index b8ee9c86c1d..c505a972891 100644 --- a/public/app/features/live/centrifuge/channel.ts +++ b/public/app/features/live/centrifuge/channel.ts @@ -1,5 +1,4 @@ import { - LiveChannelConfig, LiveChannelStatusEvent, LiveChannelEvent, LiveChannelEventType, @@ -7,6 +6,7 @@ import { LiveChannelPresenceStatus, LiveChannelAddress, DataFrameJSON, + isValidLiveChannelAddress, } from '@grafana/data'; import Centrifuge, { JoinLeaveContext, @@ -34,10 +34,9 @@ export class CentrifugeLiveChannel { // Hold on to the last header with schema lastMessageWithSchema?: DataFrameJSON; - /** Static definition of the channel definition. This may describe the channel usage */ - config?: LiveChannelConfig; subscription?: Centrifuge.Subscription; shutdownCallback?: () => void; + initalized?: boolean; constructor(id: string, addr: LiveChannelAddress) { this.id = id; @@ -48,14 +47,18 @@ export class CentrifugeLiveChannel { timestamp: this.opened, state: LiveChannelConnectionState.Pending, }; + if (!isValidLiveChannelAddress(addr)) { + this.currentStatus.state = LiveChannelConnectionState.Invalid; + this.currentStatus.error = 'invalid channel address'; + } } // This should only be called when centrifuge is connected - initalize(config: LiveChannelConfig): SubscriptionEvents { - if (this.config) { + initalize(): SubscriptionEvents { + if (this.initalized) { throw new Error('Channel already initalized: ' + this.id); } - this.config = config; + this.initalized = true; const events: SubscriptionEvents = { // Called when a message is recieved from the socket @@ -108,14 +111,12 @@ export class CentrifugeLiveChannel { }, }; - if (config.hasPresence) { - events.join = (ctx: JoinLeaveContext) => { - this.stream.next({ type: LiveChannelEventType.Join, user: ctx.info.user }); - }; - events.leave = (ctx: JoinLeaveContext) => { - this.stream.next({ type: LiveChannelEventType.Leave, user: ctx.info.user }); - }; - } + events.join = (ctx: JoinLeaveContext) => { + this.stream.next({ type: LiveChannelEventType.Join, user: ctx.info.user }); + }; + events.leave = (ctx: JoinLeaveContext) => { + this.stream.next({ type: LiveChannelEventType.Leave, user: ctx.info.user }); + }; return events; } diff --git a/public/app/features/live/centrifuge/service.ts b/public/app/features/live/centrifuge/service.ts index 71e35022dcc..1fa4ee9b1d7 100644 --- a/public/app/features/live/centrifuge/service.ts +++ b/public/app/features/live/centrifuge/service.ts @@ -10,7 +10,6 @@ import { isLiveChannelMessageEvent, isLiveChannelStatusEvent, LiveChannelAddress, - LiveChannelConfig, LiveChannelConnectionState, LiveChannelEvent, LiveChannelPresenceStatus, @@ -38,19 +37,19 @@ export interface CentrifugeSrv { /** * Watch for messages in a channel */ - getStream(address: LiveChannelAddress, config: LiveChannelConfig): Observable>; + getStream(address: LiveChannelAddress): Observable>; /** * Connect to a channel and return results as DataFrames */ - getDataStream(options: LiveDataStreamOptions, config: LiveChannelConfig): Observable; + getDataStream(options: LiveDataStreamOptions): Observable; /** * For channels that support presence, this will request the current state from the server. * * Join and leave messages will be sent to the open stream */ - getPresence(address: LiveChannelAddress, config: LiveChannelConfig): Promise; + getPresence(address: LiveChannelAddress): Promise; } export class CentrifugeService implements CentrifugeSrv { @@ -110,7 +109,7 @@ export class CentrifugeService implements CentrifugeSrv { * 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 */ - private getChannel(addr: LiveChannelAddress, config: LiveChannelConfig): CentrifugeLiveChannel { + private getChannel(addr: LiveChannelAddress): CentrifugeLiveChannel { const id = `${this.deps.orgId}/${addr.scope}/${addr.namespace}/${addr.path}`; let channel = this.open.get(id); if (channel != null) { @@ -118,13 +117,16 @@ export class CentrifugeService implements CentrifugeSrv { } channel = new CentrifugeLiveChannel(id, addr); + if (channel.currentStatus.state === LiveChannelConnectionState.Invalid) { + return channel; + } channel.shutdownCallback = () => { this.open.delete(id); // remove it from the list of open channels }; this.open.set(id, channel); // Initialize the channel in the background - this.initChannel(config, channel).catch((err) => { + this.initChannel(channel).catch((err) => { if (channel) { channel.currentStatus.state = LiveChannelConnectionState.Invalid; channel.shutdownWithError(err); @@ -136,8 +138,8 @@ export class CentrifugeService implements CentrifugeSrv { return channel; } - private async initChannel(config: LiveChannelConfig, channel: CentrifugeLiveChannel): Promise { - const events = channel.initalize(config); + private async initChannel(channel: CentrifugeLiveChannel): Promise { + const events = channel.initalize(); if (!this.centrifuge.isConnected()) { await this.connectionBlocker; } @@ -159,16 +161,16 @@ export class CentrifugeService implements CentrifugeSrv { /** * Watch for messages in a channel */ - getStream(address: LiveChannelAddress, config: LiveChannelConfig): Observable> { - return this.getChannel(address, config).getStream(); + getStream(address: LiveChannelAddress): Observable> { + return this.getChannel(address).getStream(); } /** * Connect to a channel and return results as DataFrames */ - getDataStream(options: LiveDataStreamOptions, config: LiveChannelConfig): Observable { + getDataStream(options: LiveDataStreamOptions): Observable { return new Observable((subscriber) => { - const channel = this.getChannel(options.addr, config); + const channel = this.getChannel(options.addr); const key = options.key ?? `xstr/${streamCounter++}`; let data: StreamingDataFrame | undefined = undefined; let filtered: DataFrame | undefined = undefined; @@ -273,8 +275,8 @@ export class CentrifugeService implements CentrifugeSrv { * * Join and leave messages will be sent to the open stream */ - getPresence(address: LiveChannelAddress, config: LiveChannelConfig): Promise { - return this.getChannel(address, config).getPresence(); + getPresence(address: LiveChannelAddress): Promise { + return this.getChannel(address).getPresence(); } } diff --git a/public/app/features/live/centrifuge/service.worker.ts b/public/app/features/live/centrifuge/service.worker.ts index e3ce9d19405..6d9a1d3705c 100644 --- a/public/app/features/live/centrifuge/service.worker.ts +++ b/public/app/features/live/centrifuge/service.worker.ts @@ -2,7 +2,7 @@ import { CentrifugeService, CentrifugeSrvDeps } from './service'; import * as comlink from 'comlink'; import './transferHandlers'; import { remoteObservableAsObservable } from './remoteObservable'; -import { LiveChannelAddress, LiveChannelConfig } from '@grafana/data'; +import { LiveChannelAddress } from '@grafana/data'; import { LiveDataStreamOptions } from '@grafana/runtime'; let centrifuge: CentrifugeService; @@ -23,16 +23,16 @@ const getConnectionState = () => { return comlink.proxy(centrifuge.getConnectionState()); }; -const getDataStream = (options: LiveDataStreamOptions, config: LiveChannelConfig) => { - return comlink.proxy(centrifuge.getDataStream(options, config)); +const getDataStream = (options: LiveDataStreamOptions) => { + return comlink.proxy(centrifuge.getDataStream(options)); }; -const getStream = (address: LiveChannelAddress, config: LiveChannelConfig) => { - return comlink.proxy(centrifuge.getStream(address, config)); +const getStream = (address: LiveChannelAddress) => { + return comlink.proxy(centrifuge.getStream(address)); }; -const getPresence = async (address: LiveChannelAddress, config: LiveChannelConfig) => { - return await centrifuge.getPresence(address, config); +const getPresence = async (address: LiveChannelAddress) => { + return await centrifuge.getPresence(address); }; const workObj = { diff --git a/public/app/features/live/centrifuge/serviceWorkerProxy.ts b/public/app/features/live/centrifuge/serviceWorkerProxy.ts index a1d5b369567..ac5290796da 100644 --- a/public/app/features/live/centrifuge/serviceWorkerProxy.ts +++ b/public/app/features/live/centrifuge/serviceWorkerProxy.ts @@ -4,7 +4,7 @@ import './transferHandlers'; import * as comlink from 'comlink'; import { asyncScheduler, Observable, observeOn } from 'rxjs'; -import { LiveChannelAddress, LiveChannelConfig, LiveChannelEvent } from '@grafana/data'; +import { LiveChannelAddress, LiveChannelEvent } from '@grafana/data'; import { promiseWithRemoteObservableAsObservable } from './remoteObservable'; import { createWorker } from './createCentrifugeServiceWorker'; @@ -20,21 +20,21 @@ export class CentrifugeServiceWorkerProxy implements CentrifugeSrv { return promiseWithRemoteObservableAsObservable(this.centrifugeWorker.getConnectionState()); }; - getDataStream: CentrifugeSrv['getDataStream'] = (options, config) => { - return promiseWithRemoteObservableAsObservable(this.centrifugeWorker.getDataStream(options, config)).pipe( + getDataStream: CentrifugeSrv['getDataStream'] = (options) => { + return promiseWithRemoteObservableAsObservable(this.centrifugeWorker.getDataStream(options)).pipe( // async scheduler splits the synchronous task of deserializing data from web worker and // consuming the message (ie. updating react component) into two to avoid blocking the event loop observeOn(asyncScheduler) ); }; - getPresence: CentrifugeSrv['getPresence'] = (address, config) => { - return this.centrifugeWorker.getPresence(address, config); + getPresence: CentrifugeSrv['getPresence'] = (address) => { + return this.centrifugeWorker.getPresence(address); }; - getStream: CentrifugeSrv['getStream'] = (address: LiveChannelAddress, config: LiveChannelConfig) => { + getStream: CentrifugeSrv['getStream'] = (address: LiveChannelAddress) => { return promiseWithRemoteObservableAsObservable( - this.centrifugeWorker.getStream(address, config) as Promise>>> + this.centrifugeWorker.getStream(address) as Promise>>> ); }; } diff --git a/public/app/features/live/channel-config/index.ts b/public/app/features/live/channel-config/index.ts deleted file mode 100644 index 2999c026f57..00000000000 --- a/public/app/features/live/channel-config/index.ts +++ /dev/null @@ -1,36 +0,0 @@ -import { LiveChannelScope, LiveChannelSupport, SelectableValue } from '@grafana/data'; -import { - grafanaLiveCoreFeatures, - GrafanaLiveDataSourceScope, - GrafanaLivePluginScope, - GrafanaLiveScope, - GrafanaLiveStreamScope, -} from './scope'; -import { GrafanaLiveChannelConfigSrv, ExistingLiveChannelScope } from './types'; - -export class GrafanaLiveChannelConfigService implements GrafanaLiveChannelConfigSrv { - private readonly scopes: Record; - - constructor() { - this.scopes = Object.freeze({ - [LiveChannelScope.Grafana]: grafanaLiveCoreFeatures, - [LiveChannelScope.DataSource]: new GrafanaLiveDataSourceScope(), - [LiveChannelScope.Plugin]: new GrafanaLivePluginScope(), - [LiveChannelScope.Stream]: new GrafanaLiveStreamScope(), - }); - } - - private getScope = (liveChannelScope: ExistingLiveChannelScope): GrafanaLiveScope => - this.scopes[liveChannelScope as LiveChannelScope]; - - doesScopeExist = (liveChannelScope: LiveChannelScope): liveChannelScope is ExistingLiveChannelScope => - Boolean(this.scopes[liveChannelScope]); - - getChannelSupport = async ( - liveChannelScope: ExistingLiveChannelScope, - namespace: string - ): Promise => this.getScope(liveChannelScope).getChannelSupport(namespace); - - getNamespaces = async (liveChannelScope: ExistingLiveChannelScope): Promise>> => - this.getScope(liveChannelScope).listNamespaces(); -} diff --git a/public/app/features/live/channel-config/scope.ts b/public/app/features/live/channel-config/scope.ts deleted file mode 100644 index 4339b718eb2..00000000000 --- a/public/app/features/live/channel-config/scope.ts +++ /dev/null @@ -1,182 +0,0 @@ -import { LiveChannelScope, LiveChannelSupport, SelectableValue } from '@grafana/data'; -import { getDataSourceSrv } from '@grafana/runtime'; -import { config } from 'app/core/config'; -import { loadPlugin } from 'app/features/plugins/PluginPage'; -import { LiveMeasurementsSupport } from '../measurements/measurementsSupport'; -import { CoreGrafanaLiveFeature } from './types'; - -export abstract class GrafanaLiveScope { - constructor(protected scope: LiveChannelScope) {} - - /** - * Load the real namespaces - */ - abstract getChannelSupport(namespace: string): Promise; - - /** - * List the possible values within this scope - */ - abstract listNamespaces(): Promise>>; -} - -class GrafanaLiveCoreScope extends GrafanaLiveScope { - readonly features = new Map(); - readonly namespaces: Array> = []; - - constructor() { - super(LiveChannelScope.Grafana); - } - - register(feature: CoreGrafanaLiveFeature) { - this.features.set(feature.name, feature.support); - this.namespaces.push({ - value: feature.name, - label: feature.name, - description: feature.description, - }); - } - - /** - * 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); - if (ds.channelSupport) { - return ds.channelSupport; - } - return new LiveMeasurementsSupport(); // default support? - } - - /** - * 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 = await 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) { - throw new Error('Unknown streaming plugin: ' + namespace); - } - if (plugin.channelSupport) { - return plugin.channelSupport; // explicit - } - throw new Error('Plugin does not support streaming: ' + namespace); - } - - /** - * 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 = await 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); - } -} - -export class GrafanaLiveStreamScope extends GrafanaLiveScope { - names?: Array>; - - constructor() { - super(LiveChannelScope.Stream); - } - - async getChannelSupport(namespace: string) { - return new LiveMeasurementsSupport(); - } - - /** - * List the possible values within this scope - */ - async listNamespaces() { - if (this.names) { - return Promise.resolve(this.names); - } - const names: Array> = []; - - // TODO!!! - - return (this.names = names); - } -} diff --git a/public/app/features/live/channel-config/types.ts b/public/app/features/live/channel-config/types.ts deleted file mode 100644 index 15fd2fabb1d..00000000000 --- a/public/app/features/live/channel-config/types.ts +++ /dev/null @@ -1,18 +0,0 @@ -import { LiveChannelScope, LiveChannelSupport, SelectableValue } from '@grafana/data'; - -export interface CoreGrafanaLiveFeature { - name: string; - support: LiveChannelSupport; - description: string; -} - -export type ExistingLiveChannelScope = LiveChannelScope & { readonly discriminator: unique symbol }; - -export interface GrafanaLiveChannelConfigSrv { - doesScopeExist: (liveChannelScope: LiveChannelScope) => liveChannelScope is ExistingLiveChannelScope; - getChannelSupport: ( - liveChannelScope: ExistingLiveChannelScope, - namespace: string - ) => Promise; - getNamespaces: (liveChannelScope: ExistingLiveChannelScope) => Promise>>; -} diff --git a/public/app/features/live/dashboard/dashboardWatcher.ts b/public/app/features/live/dashboard/dashboardWatcher.ts index 7ab8dc8b6d3..f4f89713845 100644 --- a/public/app/features/live/dashboard/dashboardWatcher.ts +++ b/public/app/features/live/dashboard/dashboardWatcher.ts @@ -12,7 +12,6 @@ import { } from '@grafana/data'; import { DashboardChangedModal } from './DashboardChangedModal'; import { DashboardEvent, DashboardEventAction } from './types'; -import { CoreGrafanaLiveFeature } from '../channel-config/types'; import { sessionId } from 'app/features/live'; import { ShowModalReactEvent } from '../../../types/events'; import { Unsubscribable } from 'rxjs'; @@ -159,16 +158,3 @@ class DashboardWatcher { } export const dashboardWatcher = new DashboardWatcher(); - -export function getDashboardChannelsFeature(): CoreGrafanaLiveFeature { - return { - name: 'dashboard', - support: { - getChannelConfig: (path: string) => ({ - description: 'Dashboard change events', - hasPresence: true, - }), - }, - description: 'Dashboard listener', - }; -} diff --git a/public/app/features/live/features.ts b/public/app/features/live/features.ts deleted file mode 100644 index 7c5323ffcf5..00000000000 --- a/public/app/features/live/features.ts +++ /dev/null @@ -1,34 +0,0 @@ -import { LiveChannelType } from '@grafana/data'; -import { getDashboardChannelsFeature } from './dashboard/dashboardWatcher'; -import { grafanaLiveCoreFeatures } from './channel-config/scope'; - -export function registerLiveFeatures() { - grafanaLiveCoreFeatures.register({ - name: 'testdata', - support: { - getChannelConfig: (path: string) => { - return { - type: LiveChannelType.DataStream, - }; - }, - }, - description: 'Test data generations', - }); - - grafanaLiveCoreFeatures.register({ - name: 'broadcast', - support: { - getChannelConfig: (path: string) => { - return { - type: LiveChannelType.JSON, - canPublish: true, - description: 'Broadcast any messages to a channel', - }; - }, - }, - description: 'Broadcast will send/receive any JSON object in a channel', - }); - - // dashboard/* - grafanaLiveCoreFeatures.register(getDashboardChannelsFeature()); -} diff --git a/public/app/features/live/index.ts b/public/app/features/live/index.ts index 12b7baca0c9..31a2001c828 100644 --- a/public/app/features/live/index.ts +++ b/public/app/features/live/index.ts @@ -1,17 +1,10 @@ import { config, getBackendSrv, getGrafanaLiveSrv, setGrafanaLiveSrv } from '@grafana/runtime'; -import { registerLiveFeatures } from './features'; import { GrafanaLiveService } from './live'; -import { GrafanaLiveChannelConfigService } from './channel-config'; -import { GrafanaLiveChannelConfigSrv } from './channel-config/types'; import { contextSrv } from '../../core/services/context_srv'; import { CentrifugeServiceWorkerProxy } from './centrifuge/serviceWorkerProxy'; import { CentrifugeService } from './centrifuge/service'; import { liveTimer } from 'app/features/dashboard/dashgrid/liveTimer'; -const grafanaLiveScopesSingleton = new GrafanaLiveChannelConfigService(); - -export const getGrafanaLiveScopes = (): GrafanaLiveChannelConfigSrv => grafanaLiveScopesSingleton; - export const sessionId = (window as any)?.grafanaBootData?.user?.id + '/' + @@ -35,12 +28,10 @@ export function initGrafanaLive() { setGrafanaLiveSrv( new GrafanaLiveService({ - scopes: getGrafanaLiveScopes(), centrifugeSrv, backendSrv: getBackendSrv(), }) ); - registerLiveFeatures(); } export function getGrafanaLiveCentrifugeSrv() { diff --git a/public/app/features/live/live.ts b/public/app/features/live/live.ts index b09606cb9d2..f9fd1afdda1 100644 --- a/public/app/features/live/live.ts +++ b/public/app/features/live/live.ts @@ -1,24 +1,16 @@ import { BackendSrv, GrafanaLiveSrv, LiveDataStreamOptions } from '@grafana/runtime'; import { CentrifugeSrv } from './centrifuge/service'; -import { mergeMap, from, of, Observable } from 'rxjs'; +import { Observable } from 'rxjs'; import { DataQueryResponse, - isValidLiveChannelAddress, LiveChannelAddress, - LiveChannelConfig, - LiveChannelConnectionState, LiveChannelEvent, - LiveChannelEventType, LiveChannelPresenceStatus, - LoadingState, toLiveChannelId, } from '@grafana/data'; -import { GrafanaLiveChannelConfigSrv } from './channel-config/types'; -import { catchError } from 'rxjs/operators'; type GrafanaLiveServiceDeps = { - scopes: GrafanaLiveChannelConfigSrv; centrifugeSrv: CentrifugeSrv; backendSrv: BackendSrv; }; @@ -37,23 +29,14 @@ export class GrafanaLiveService implements GrafanaLiveSrv { * Connect to a channel and return results as DataFrames */ getDataStream(options: LiveDataStreamOptions): Observable { - const channelConfig = this.getChannelInfo(options.addr); - - return from(channelConfig).pipe( - mergeMap((config) => this.deps.centrifugeSrv.getDataStream(options, config)), - catchError((error) => this.getInvalidDataStream(error, options)) - ); + return this.deps.centrifugeSrv.getDataStream(options); } /** * Watch for messages in a channel */ getStream(address: LiveChannelAddress): Observable> { - const channelConfig = this.getChannelInfo(address); - return from(channelConfig).pipe( - mergeMap((config) => this.deps.centrifugeSrv.getStream(address, config)), - catchError((error) => this.getInvalidChannelStream(error, address)) - ); + return this.deps.centrifugeSrv.getStream(address); } /** @@ -74,55 +57,6 @@ export class GrafanaLiveService implements GrafanaLiveSrv { * Join and leave messages will be sent to the open stream */ async getPresence(address: LiveChannelAddress): Promise { - const channelConfig = await this.getChannelInfo(address); - return this.deps.centrifugeSrv.getPresence(address, channelConfig); + return this.deps.centrifugeSrv.getPresence(address); } - - /** - * 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 - */ - async getChannelInfo(addr: LiveChannelAddress): Promise { - if (!isValidLiveChannelAddress(addr)) { - return Promise.reject('invalid live channel address'); - } - - if (!this.deps.scopes.doesScopeExist(addr.scope)) { - return Promise.reject('invalid scope'); - } - - const support = await this.deps.scopes.getChannelSupport(addr.scope, addr.namespace); - if (!support) { - return Promise.reject(addr.namespace + ' does not support streaming'); - } - return support.getChannelConfig(addr.path)!; - } - - private getInvalidChannelStream = (error: Error, address: LiveChannelAddress): Observable> => { - return of({ - type: LiveChannelEventType.Status, - id: `${address.scope}/${address.namespace}/${address.path}`, - timestamp: Date.now(), - state: LiveChannelConnectionState.Invalid, - error, - message: error.message, - }); - }; - - private getInvalidDataStream = (error: Error, options: LiveDataStreamOptions): Observable => { - return of({ - error: { - data: { - error: error.stack, - }, - message: error.message, - }, - state: LoadingState.Error, - data: options.frame ? [options.frame] : [], - }); - }; } diff --git a/public/app/features/live/measurements/measurementsSupport.ts b/public/app/features/live/measurements/measurementsSupport.ts deleted file mode 100644 index e7115b7565b..00000000000 --- a/public/app/features/live/measurements/measurementsSupport.ts +++ /dev/null @@ -1,17 +0,0 @@ -import { LiveChannelSupport, LiveChannelConfig, LiveChannelType } from '@grafana/data'; - -/** - * Generic description of channels that support streams - * - * @alpha - */ -export class LiveMeasurementsSupport implements LiveChannelSupport { - /** - * Get the channel handler for the path, or throw an error if invalid - */ - getChannelConfig(path: string): LiveChannelConfig | undefined { - return { - type: LiveChannelType.DataStream, - }; - } -} diff --git a/public/app/plugins/datasource/cloudwatch/module.tsx b/public/app/plugins/datasource/cloudwatch/module.tsx index d35e8597f70..da769501722 100644 --- a/public/app/plugins/datasource/cloudwatch/module.tsx +++ b/public/app/plugins/datasource/cloudwatch/module.tsx @@ -6,7 +6,6 @@ import { CloudWatchJsonData, CloudWatchQuery } from './types'; import { CloudWatchLogsQueryEditor } from './components/LogsQueryEditor'; import { PanelQueryEditor } from './components/PanelQueryEditor'; import LogsCheatSheet from './components/LogsCheatSheet'; -import { LiveMeasurementsSupport } from 'app/features/live/measurements/measurementsSupport'; export const plugin = new DataSourcePlugin( CloudWatchDatasource @@ -16,5 +15,4 @@ export const plugin = new DataSourcePlugin; @@ -25,7 +23,6 @@ const scopes: Array> = [ interface State { namespaces: Array>; paths: Array>; - support?: LiveChannelSupport; } export class LiveChannelEditor extends PureComponent { @@ -44,31 +41,9 @@ export class LiveChannelEditor extends PureComponent { } } - async getScopeDetails() { - const { scope, namespace } = this.props.value; - const srv = getGrafanaLiveScopes(); - - if (!srv.doesScopeExist(scope)) { - return { - namespaces: [], - support: undefined, - }; - } - - const namespaces = await srv.getNamespaces(scope); - const support = namespace ? await srv.getChannelSupport(scope, namespace) : undefined; - return { - namespaces, - support, - }; - } - async updateSelectOptions() { - const { namespaces, support } = await this.getScopeDetails(); - this.setState({ - namespaces, - support, + namespaces: [], paths: [], }); }