Live: move centrifuge service to a web worker (#41090)

* Fix: make webpack pickup workers written in TS

* Add comlink to dependencies

* Temporary fix: copy paste `toDataQueryError` from @grafana/runtime to avoid web dependencies

* Implemented comlink-based centrifuge worker & worker proxy

* Temporary fix: implement comlink transferHandlers for subscriptions and streamingdataframes

* Move liveTimer filtering from CentrifugeService into GrafanaLiveService

* Switch from CentrifugeService to CentrifugeServiceWorkerProxy in GrafanaLive

* Naming fix

* Refactor: move liveTimer-based data filtering from GrafanaLiveService to CentrifugeServiceWorker

* observe dataStream on an async scheduler

* Fix: - Unsubscribe is now propagated from the main thread to the worker, - improve worker&workerProxy types

* Fix: Prettify types

* Fix: Add error & complete observers

* Docs: Add comment explaining the `subscriberTransferHandler`

* Fix: Replace `StreamingDataFrameHandler` with explicitly converting StreamingDataFrame to a DataFrameDTO

* Refactor: move liveTimer filtering to service.ts to make it easy to implement a `live-service-web-worker` feature flag

* Feat: add `live-service-web-worker` feature flag

* Fix: extract toDataQueryError.ts to a separate file within `@grafana-runtime` to avoid having a dependency from webworker to the whole package (@grafana-runtime/index.ts)

* Update public/app/features/dashboard/dashgrid/liveTimer.ts

Co-authored-by: Leon Sorokin <leeoniya@gmail.com>

* Fix: fixed default import class in worker file

* Fix: cast worker as Endpoint

* Migrate from worker-loader to webpack native worker support v1 - broken prod build

* Fix: Use custom path in HtmlWebpackPlugin

* Fix: Loading workers from CDNs

* Fix: Avoid issues with jest ESM support by mocking `createWorker` files

* Fix: move the custom mockWorker rendering layout to `test/mocks`

Co-authored-by: Leon Sorokin <leeoniya@gmail.com>

public/app/features/live/centrifuge/createCentrifugeServiceWorker.ts

@@ -0,0 +1,3 @@
+import { CorsWorker as Worker } from 'app/core/utils/CorsWorker';
+
+export const createWorker = () => new Worker(new URL('./service.worker.ts', import.meta.url));

public/app/features/live/centrifuge/remoteObservable.ts

@@ -0,0 +1,33 @@
+import * as comlink from 'comlink';
+import { from, Observable, switchMap } from 'rxjs';
+
+export const remoteObservableAsObservable = <T>(remoteObs: comlink.RemoteObject<Observable<T>>): Observable<T> =>
+  new Observable((subscriber) => {
+    // Passing the callbacks as 3 separate arguments is deprecated, but it's the only option for now
+    //
+    // RxJS recreates the functions via `Function.bind` https://github.com/ReactiveX/rxjs/blob/62aca850a37f598b5db6085661e0594b81ec4281/src/internal/Subscriber.ts#L169
+    // and thus erases the ProxyMarker created via comlink.proxy(fN) when the callbacks
+    // are grouped together in a Observer object (ie. { next: (v) => ..., error: (err) => ..., complete: () => ... })
+    //
+    // solution: TBD (autoproxy all functions?)
+    const remoteSubPromise = remoteObs.subscribe(
+      comlink.proxy((nextValueInRemoteObs: T) => {
+        subscriber.next(nextValueInRemoteObs);
+      }),
+      comlink.proxy((err) => {
+        subscriber.error(err);
+      }),
+      comlink.proxy(() => {
+        subscriber.complete();
+      })
+    );
+    return {
+      unsubscribe: () => {
+        remoteSubPromise.then((remoteSub) => remoteSub.unsubscribe());
+      },
+    };
+  });
+
+export const promiseWithRemoteObservableAsObservable = <T>(
+  promiseWithProxyObservable: Promise<comlink.RemoteObject<Observable<T>>>
+): Observable<T> => from(promiseWithProxyObservable).pipe(switchMap((val) => remoteObservableAsObservable(val)));

public/app/features/live/centrifuge/service.ts

@@ -1,5 +1,6 @@
 import Centrifuge from 'centrifuge/dist/centrifuge';
-import { LiveDataStreamOptions, toDataQueryError } from '@grafana/runtime';
+import { LiveDataStreamOptions } from '@grafana/runtime';
+import { toDataQueryError } from '@grafana/runtime/src/utils/toDataQueryError';
 import { BehaviorSubject, Observable } from 'rxjs';
 import {
   DataFrame,
@@ -15,25 +16,52 @@ import {
   LiveChannelPresenceStatus,
   LoadingState,
   StreamingDataFrame,
+  toDataFrameDTO,
 } from '@grafana/data';
 import { CentrifugeLiveChannel } from './channel';
-import { liveTimer } from 'app/features/dashboard/dashgrid/liveTimer';
 
-type CentrifugeSrvDeps = {
+export type CentrifugeSrvDeps = {
   appUrl: string;
   orgId: number;
   orgRole: string;
   sessionId: string;
   liveEnabled: boolean;
+  dataStreamSubscriberReadiness: Observable<boolean>;
 };
 
-export class CentrifugeSrv {
+export interface CentrifugeSrv {
+  /**
+   * Listen for changes to the connection state
+   */
+  getConnectionState(): Observable<boolean>;
+
+  /**
+   * Watch for messages in a channel
+   */
+  getStream<T>(address: LiveChannelAddress, config: LiveChannelConfig): Observable<LiveChannelEvent<T>>;
+
+  /**
+   * Connect to a channel and return results as DataFrames
+   */
+  getDataStream(options: LiveDataStreamOptions, config: LiveChannelConfig): 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>;
+}
+
+export class CentrifugeService implements CentrifugeSrv {
   readonly open = new Map<string, CentrifugeLiveChannel>();
   readonly centrifuge: Centrifuge;
   readonly connectionState: BehaviorSubject<boolean>;
   readonly connectionBlocker: Promise<void>;
+  private dataStreamSubscriberReady = true;
 
   constructor(private deps: CentrifugeSrvDeps) {
+    deps.dataStreamSubscriberReadiness.subscribe((next) => (this.dataStreamSubscriberReady = next));
     const liveUrl = `${deps.appUrl.replace(/^http/, 'ws')}/api/live/ws`;
     this.centrifuge = new Centrifuge(liveUrl, {});
     this.centrifuge.setConnectData({
@@ -66,15 +94,15 @@ export class CentrifugeSrv {
   // Internal functions
   //----------------------------------------------------------
 
-  onConnect = (context: any) => {
+  private onConnect = (context: any) => {
     this.connectionState.next(true);
   };
 
-  onDisconnect = (context: any) => {
+  private onDisconnect = (context: any) => {
     this.connectionState.next(false);
   };
 
-  onServerSideMessage = (context: any) => {
+  private onServerSideMessage = (context: any) => {
     console.log('Publication from server-side channel', context);
   };
 
@@ -82,7 +110,7 @@ export class 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
    */
-  getChannel<TMessage>(addr: LiveChannelAddress, config: LiveChannelConfig): CentrifugeLiveChannel<TMessage> {
+  private getChannel<TMessage>(addr: LiveChannelAddress, config: LiveChannelConfig): CentrifugeLiveChannel<TMessage> {
     const id = `${this.deps.orgId}/${addr.scope}/${addr.namespace}/${addr.path}`;
     let channel = this.open.get(id);
     if (channel != null) {
@@ -145,7 +173,6 @@ export class CentrifugeSrv {
       let data: StreamingDataFrame | undefined = undefined;
       let filtered: DataFrame | undefined = undefined;
       let state = LoadingState.Streaming;
-      let last = liveTimer.lastUpdate;
       let lastWidth = -1;
 
       const process = (msg: DataFrameJSON) => {
@@ -172,11 +199,18 @@ export class CentrifugeSrv {
           }
         }
 
-        const elapsed = liveTimer.lastUpdate - last;
-        if (elapsed > 1000 || liveTimer.ok) {
+        if (this.dataStreamSubscriberReady) {
           filtered.length = data.length; // make sure they stay up-to-date
-          subscriber.next({ state, data: [filtered], key });
-          last = liveTimer.lastUpdate;
+          subscriber.next({
+            state,
+            data: [
+              // workaround for serializing issues when sending DataFrame from web worker to the main thread
+              // DataFrame is making use of ArrayVectors which are es6 classes and thus not cloneable out of the box
+              // `toDataFrameDTO` converts ArrayVectors into native arrays.
+              toDataFrameDTO(filtered),
+            ],
+            key,
+          });
         }
       };
 

public/app/features/live/centrifuge/service.worker.ts

@@ -0,0 +1,52 @@
+import { CentrifugeService, CentrifugeSrvDeps } from './service';
+import * as comlink from 'comlink';
+import './transferHandlers';
+import { remoteObservableAsObservable } from './remoteObservable';
+import { LiveChannelAddress, LiveChannelConfig } from '@grafana/data';
+import { LiveDataStreamOptions } from '@grafana/runtime';
+
+let centrifuge: CentrifugeService;
+
+const initialize = (
+  deps: CentrifugeSrvDeps,
+  remoteDataStreamSubscriberReadiness: comlink.RemoteObject<
+    CentrifugeSrvDeps['dataStreamSubscriberReadiness'] & comlink.ProxyMarked
+  >
+) => {
+  centrifuge = new CentrifugeService({
+    ...deps,
+    dataStreamSubscriberReadiness: remoteObservableAsObservable(remoteDataStreamSubscriberReadiness),
+  });
+};
+
+const getConnectionState = () => {
+  return comlink.proxy(centrifuge.getConnectionState());
+};
+
+const getDataStream = (options: LiveDataStreamOptions, config: LiveChannelConfig) => {
+  return comlink.proxy(centrifuge.getDataStream(options, config));
+};
+
+const getStream = (address: LiveChannelAddress, config: LiveChannelConfig) => {
+  return comlink.proxy(centrifuge.getStream(address, config));
+};
+
+const getPresence = async (address: LiveChannelAddress, config: LiveChannelConfig) => {
+  return await centrifuge.getPresence(address, config);
+};
+
+const workObj = {
+  initialize,
+  getConnectionState,
+  getDataStream,
+  getStream,
+  getPresence,
+};
+
+export type RemoteCentrifugeService = typeof workObj;
+
+comlink.expose(workObj);
+
+export default class {
+  constructor() {}
+}

public/app/features/live/centrifuge/serviceWorkerProxy.ts

@@ -0,0 +1,40 @@
+import { CentrifugeSrv, CentrifugeSrvDeps } from './service';
+import { RemoteCentrifugeService } from './service.worker';
+import './transferHandlers';
+
+import * as comlink from 'comlink';
+import { asyncScheduler, Observable, observeOn } from 'rxjs';
+import { LiveChannelAddress, LiveChannelConfig, LiveChannelEvent } from '@grafana/data';
+import { promiseWithRemoteObservableAsObservable } from './remoteObservable';
+import { createWorker } from './createCentrifugeServiceWorker';
+
+export class CentrifugeServiceWorkerProxy implements CentrifugeSrv {
+  private centrifugeWorker;
+
+  constructor(deps: CentrifugeSrvDeps) {
+    this.centrifugeWorker = comlink.wrap<RemoteCentrifugeService>(createWorker() as comlink.Endpoint);
+    this.centrifugeWorker.initialize(deps, comlink.proxy(deps.dataStreamSubscriberReadiness));
+  }
+
+  getConnectionState: CentrifugeSrv['getConnectionState'] = () => {
+    return promiseWithRemoteObservableAsObservable(this.centrifugeWorker.getConnectionState());
+  };
+
+  getDataStream: CentrifugeSrv['getDataStream'] = (options, config) => {
+    return promiseWithRemoteObservableAsObservable(this.centrifugeWorker.getDataStream(options, config)).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);
+  };
+
+  getStream: CentrifugeSrv['getStream'] = <T>(address: LiveChannelAddress, config: LiveChannelConfig) => {
+    return promiseWithRemoteObservableAsObservable(
+      this.centrifugeWorker.getStream(address, config) as Promise<comlink.Remote<Observable<LiveChannelEvent<T>>>>
+    );
+  };
+}

public/app/features/live/centrifuge/transferHandlers.ts

@@ -0,0 +1,27 @@
+import * as comlink from 'comlink';
+import { Subscriber } from 'rxjs';
+
+// Observers, ie. functions passed to `observable.subscribe(...)`, are converted to a subclass of `Subscriber` before they are sent to the source Observable.
+// The conversion happens internally in the RxJS library - this transfer handler is catches them and wraps them with a proxy
+const subscriberTransferHandler: any = {
+  canHandle(value: any): boolean {
+    return value && value instanceof Subscriber;
+  },
+
+  serialize(value: Function): [MessagePort, Transferable[]] {
+    const obj = comlink.proxy(value);
+
+    const { port1, port2 } = new MessageChannel();
+
+    comlink.expose(obj, port1);
+
+    return [port2, [port2]];
+  },
+
+  deserialize(value: MessagePort): comlink.Remote<MessagePort> {
+    value.start();
+
+    return comlink.wrap<MessagePort>(value);
+  },
+};
+comlink.transferHandlers.set('SubscriberHandler', subscriberTransferHandler);

public/app/features/live/index.ts

@@ -1,10 +1,12 @@
 import { config, getBackendSrv, getGrafanaLiveSrv, setGrafanaLiveSrv } from '@grafana/runtime';
-import { CentrifugeSrv } from './centrifuge/service';
 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();
 
@@ -18,13 +20,19 @@ export const sessionId =
   Math.random().toString(36).substring(2, 15);
 
 export function initGrafanaLive() {
-  const centrifugeSrv = new CentrifugeSrv({
+  const centrifugeServiceDeps = {
     appUrl: `${window.location.origin}${config.appSubUrl}`,
     orgId: contextSrv.user.orgId,
     orgRole: contextSrv.user.orgRole,
     liveEnabled: config.liveEnabled,
     sessionId,
-  });
+    dataStreamSubscriberReadiness: liveTimer.ok.asObservable(),
+  };
+
+  const centrifugeSrv = config.featureToggles['live-service-web-worker']
+    ? new CentrifugeServiceWorkerProxy(centrifugeServiceDeps)
+    : new CentrifugeService(centrifugeServiceDeps);
+
   setGrafanaLiveSrv(
     new GrafanaLiveService({
       scopes: getGrafanaLiveScopes(),