Live: remove hardcoded frontend channel support (#41584)

This commit is contained in:
Ryan McKinley
2021-11-15 07:54:14 -08:00
committed by GitHub
parent bfec79a7dc
commit c780854a18
19 changed files with 54 additions and 520 deletions
+15 -14
View File
@@ -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<T = any> {
// 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<T = any> {
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<T = any> {
},
};
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;
}
+16 -14
View File
@@ -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<T>(address: LiveChannelAddress, config: LiveChannelConfig): Observable<LiveChannelEvent<T>>;
getStream<T>(address: LiveChannelAddress): Observable<LiveChannelEvent<T>>;
/**
* Connect to a channel and return results as DataFrames
*/
getDataStream(options: LiveDataStreamOptions, config: LiveChannelConfig): Observable<DataQueryResponse>;
getDataStream(options: LiveDataStreamOptions): Observable<DataQueryResponse>;
/**
* 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<LiveChannelPresenceStatus>;
getPresence(address: LiveChannelAddress): Promise<LiveChannelPresenceStatus>;
}
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<TMessage>(addr: LiveChannelAddress, config: LiveChannelConfig): CentrifugeLiveChannel<TMessage> {
private getChannel<TMessage>(addr: LiveChannelAddress): CentrifugeLiveChannel<TMessage> {
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<void> {
const events = channel.initalize(config);
private async initChannel(channel: CentrifugeLiveChannel): Promise<void> {
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<T>(address: LiveChannelAddress, config: LiveChannelConfig): Observable<LiveChannelEvent<T>> {
return this.getChannel<T>(address, config).getStream();
getStream<T>(address: LiveChannelAddress): Observable<LiveChannelEvent<T>> {
return this.getChannel<T>(address).getStream();
}
/**
* Connect to a channel and return results as DataFrames
*/
getDataStream(options: LiveDataStreamOptions, config: LiveChannelConfig): Observable<DataQueryResponse> {
getDataStream(options: LiveDataStreamOptions): Observable<DataQueryResponse> {
return new Observable<DataQueryResponse>((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<LiveChannelPresenceStatus> {
return this.getChannel(address, config).getPresence();
getPresence(address: LiveChannelAddress): Promise<LiveChannelPresenceStatus> {
return this.getChannel(address).getPresence();
}
}
@@ -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 = {
@@ -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'] = <T>(address: LiveChannelAddress, config: LiveChannelConfig) => {
getStream: CentrifugeSrv['getStream'] = <T>(address: LiveChannelAddress) => {
return promiseWithRemoteObservableAsObservable(
this.centrifugeWorker.getStream(address, config) as Promise<comlink.Remote<Observable<LiveChannelEvent<T>>>>
this.centrifugeWorker.getStream(address) as Promise<comlink.Remote<Observable<LiveChannelEvent<T>>>>
);
};
}
@@ -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<LiveChannelScope, GrafanaLiveScope>;
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<LiveChannelSupport | undefined> => this.getScope(liveChannelScope).getChannelSupport(namespace);
getNamespaces = async (liveChannelScope: ExistingLiveChannelScope): Promise<Array<SelectableValue<string>>> =>
this.getScope(liveChannelScope).listNamespaces();
}
@@ -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<LiveChannelSupport | undefined>;
/**
* List the possible values within this scope
*/
abstract listNamespaces(): Promise<Array<SelectableValue<string>>>;
}
class GrafanaLiveCoreScope extends GrafanaLiveScope {
readonly features = new Map<string, LiveChannelSupport>();
readonly namespaces: Array<SelectableValue<string>> = [];
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<SelectableValue<string>>;
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<SelectableValue<string>> = [];
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<SelectableValue<string>>;
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<SelectableValue<string>> = [];
// 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<SelectableValue<string>>;
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<SelectableValue<string>> = [];
// TODO!!!
return (this.names = names);
}
}
@@ -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<LiveChannelSupport | undefined>;
getNamespaces: (liveChannelScope: ExistingLiveChannelScope) => Promise<Array<SelectableValue<string>>>;
}
@@ -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',
};
}
-34
View File
@@ -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());
}
-9
View File
@@ -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() {
+4 -70
View File
@@ -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<DataQueryResponse> {
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<T>(address: LiveChannelAddress): Observable<LiveChannelEvent<T>> {
const channelConfig = this.getChannelInfo(address);
return from(channelConfig).pipe(
mergeMap((config) => this.deps.centrifugeSrv.getStream<T>(address, config)),
catchError((error) => this.getInvalidChannelStream<T>(error, address))
);
return this.deps.centrifugeSrv.getStream<T>(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<LiveChannelPresenceStatus> {
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<LiveChannelConfig> {
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 = <T>(error: Error, address: LiveChannelAddress): Observable<LiveChannelEvent<T>> => {
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<DataQueryResponse> => {
return of({
error: {
data: {
error: error.stack,
},
message: error.message,
},
state: LoadingState.Error,
data: options.frame ? [options.frame] : [],
});
};
}
@@ -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,
};
}
}