Live: cleanup and simple changes (#28028)

devenv/docker/blocks/influxdb/telegraf.conf

@@ -3346,7 +3346,7 @@
 #   # pid_finder = "pgrep"
 
 
-# # Reads last_run_summary.yaml file and converts to measurments
+# # Reads last_run_summary.yaml file and converts to measurements
 # [[inputs.puppetagent]]
 #   ## Location of puppet last run summary file
 #   location = "/var/lib/puppet/state/last_run_summary.yaml"

packages/grafana-data/src/types/live.ts

@@ -146,18 +146,21 @@ export interface LiveChannelPresenceStatus {
 /**
  * @experimental
  */
+export interface LiveChannelAddress {
+  scope: LiveChannelScope;
+  namespace: string; // depends on the scope
+  path: string;
+}
+
+/**
+ * @experimental
+ */
 export interface LiveChannel<TMessage = any, TPublish = any> {
   /** 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;
+  /** The channel address */
+  addr: LiveChannelAddress;
 
   /** Unix timestamp for when the channel connected */
   opened: number;

packages/grafana-data/src/utils/labels.test.ts

-import { parseLabels, formatLabels, findCommonLabels, findUniqueLabels } from './labels';
+import { parseLabels, formatLabels, findCommonLabels, findUniqueLabels, matchAllLabels } from './labels';
+import { Labels } from '../types/data';
 
 describe('parseLabels()', () => {
   it('returns no labels on empty labels string', () => {
@@ -53,3 +54,25 @@ describe('findUniqueLabels()', () => {
     expect(findUniqueLabels({ foo: '"bar"', baz: '"42"' }, { foo: '"bar"' })).toEqual({ baz: '"42"' });
   });
 });
+
+describe('matchAllLabels()', () => {
+  it('empty labels do math', () => {
+    expect(matchAllLabels({}, {})).toBeTruthy();
+  });
+
+  it('missing labels', () => {
+    expect(matchAllLabels({ foo: 'bar' }, {})).toBeFalsy();
+  });
+
+  it('extra labels should match', () => {
+    expect(matchAllLabels({ foo: 'bar' }, { foo: 'bar', baz: '22' })).toBeTruthy();
+  });
+
+  it('be graceful with null values (match)', () => {
+    expect(matchAllLabels({ foo: 'bar' })).toBeFalsy();
+  });
+
+  it('be graceful with null values (match)', () => {
+    expect(matchAllLabels((undefined as unknown) as Labels, { foo: 'bar' })).toBeTruthy();
+  });
+});

packages/grafana-data/src/utils/labels.ts

@@ -60,6 +60,21 @@ export function findUniqueLabels(labels: Labels | undefined, commonLabels: Label
 }
 
 /**
+ * Check that all labels exist in another set of labels
+ */
+export function matchAllLabels(expect: Labels, against?: Labels): boolean {
+  if (!expect) {
+    return true; // nothing to match
+  }
+  for (const [key, value] of Object.entries(expect)) {
+    if (!against || against[key] !== value) {
+      return false;
+    }
+  }
+  return true;
+}
+
+/**
  * Serializes the given labels to a string.
  */
 export function formatLabels(labels: Labels, defaultValue = '', withoutBraces?: boolean): string {

packages/grafana-runtime/src/services/live.ts

-import { LiveChannel, LiveChannelScope } from '@grafana/data';
+import { LiveChannel, LiveChannelAddress } from '@grafana/data';
 import { Observable } from 'rxjs';
 
 /**
@@ -23,11 +23,7 @@ export interface GrafanaLiveSrv {
    * Multiple requests for this channel will return the same object until
    * the channel is shutdown
    */
-  getChannel<TMessage, TPublish>(
-    scope: LiveChannelScope,
-    namespace: string,
-    path: string
-  ): LiveChannel<TMessage, TPublish>;
+  getChannel<TMessage, TPublish = any>(address: LiveChannelAddress): LiveChannel<TMessage, TPublish>;
 }
 
 let singletonInstance: GrafanaLiveSrv;

pkg/api/api.go

@@ -426,7 +426,7 @@ func (hs *HTTPServer) registerRoutes() {
 
 	// Live streaming
 	if hs.Live != nil {
-		r.Any("/live/*", hs.Live.Handler)
+		r.Any("/live/*", hs.Live.WebsocketHandler)
 	}
 
 	// Snapshots

pkg/services/live/live.go

@@ -31,7 +31,7 @@ type GrafanaLive struct {
 	node *centrifuge.Node
 
 	// The websocket handler
-	Handler interface{}
+	WebsocketHandler interface{}
 
 	// Full channel handler
 	channels   map[string]models.ChannelHandler
@@ -171,7 +171,7 @@ func InitializeBroker() (*GrafanaLive, error) {
 		WriteBufferSize: 1024,
 	})
 
-	glive.Handler = func(ctx *models.ReqContext) {
+	glive.WebsocketHandler = func(ctx *models.ReqContext) {
 		user := ctx.SignedInUser
 		if user == nil {
 			ctx.Resp.WriteHeader(401)

public/app/features/admin/LivePanel.tsx

@@ -62,7 +62,7 @@ export class LivePanel extends PureComponent<Props, State> {
 
   startSubscription = () => {
     const { scope, namespace, path } = this.props;
-    const channel = getGrafanaLiveSrv().getChannel(scope, namespace, path);
+    const channel = getGrafanaLiveSrv().getChannel({ scope, namespace, path });
     if (this.state.channel === channel) {
       return; // no change!
     }

public/app/features/live/channel.ts

 import {
   LiveChannelConfig,
   LiveChannel,
-  LiveChannelScope,
   LiveChannelStatusEvent,
   LiveChannelEvent,
   LiveChannelEventType,
   LiveChannelConnectionState,
   LiveChannelPresenceStatus,
+  LiveChannelAddress,
 } from '@grafana/data';
 import Centrifuge, {
   JoinLeaveContext,
@@ -26,9 +26,7 @@ export class CentrifugeLiveChannel<TMessage = any, TPublish = any> implements Li
 
   readonly opened = Date.now();
   readonly id: string;
-  readonly scope: LiveChannelScope;
-  readonly namespace: string;
-  readonly path: string;
+  readonly addr: LiveChannelAddress;
 
   readonly stream = new Subject<LiveChannelEvent<TMessage>>();
 
@@ -37,11 +35,9 @@ export class CentrifugeLiveChannel<TMessage = any, TPublish = any> implements Li
   subscription?: Centrifuge.Subscription;
   shutdownCallback?: () => void;
 
-  constructor(id: string, scope: LiveChannelScope, namespace: string, path: string) {
+  constructor(id: string, addr: LiveChannelAddress) {
     this.id = id;
-    this.scope = scope;
-    this.namespace = namespace;
-    this.path = path;
+    this.addr = addr;
     this.currentStatus = {
       type: LiveChannelEventType.Status,
       id,
@@ -61,15 +57,25 @@ export class CentrifugeLiveChannel<TMessage = any, TPublish = any> implements Li
     const events: SubscriptionEvents = {
       // This means a message was received from the server
       publish: (ctx: PublicationContext) => {
-        this.stream.next({
-          type: LiveChannelEventType.Message,
-          message: prepare(ctx.data),
-        });
-
-        // Clear any error messages
-        if (this.currentStatus.error) {
+        try {
+          const message = prepare(ctx.data);
+          if (message) {
+            this.stream.next({
+              type: LiveChannelEventType.Message,
+              message,
+            });
+          }
+
+          // Clear any error messages
+          if (this.currentStatus.error) {
+            this.currentStatus.timestamp = Date.now();
+            delete this.currentStatus.error;
+            this.sendStatus();
+          }
+        } catch (err) {
+          console.log('publish error', config.path, err);
+          this.currentStatus.error = err;
           this.currentStatus.timestamp = Date.now();
-          delete this.currentStatus.error;
           this.sendStatus();
         }
       },
@@ -81,6 +87,7 @@ export class CentrifugeLiveChannel<TMessage = any, TPublish = any> implements Li
       subscribe: (ctx: SubscribeSuccessContext) => {
         this.currentStatus.timestamp = Date.now();
         this.currentStatus.state = LiveChannelConnectionState.Connected;
+        delete this.currentStatus.error;
         this.sendStatus();
       },
       unsubscribe: (ctx: UnsubscribeContext) => {
@@ -159,19 +166,11 @@ export class CentrifugeLiveChannel<TMessage = any, TPublish = any> implements Li
   }
 }
 
-export function getErrorChannel(
-  msg: string,
-  id: string,
-  scope: LiveChannelScope,
-  namespace: string,
-  path: string
-): LiveChannel {
+export function getErrorChannel(msg: string, id: string, addr: LiveChannelAddress): LiveChannel {
   return {
     id,
     opened: Date.now(),
-    scope,
-    namespace,
-    path,
+    addr,
 
     // return an error
     getStream: () =>

public/app/features/live/dashboard/dashboardWatcher.ts

@@ -56,7 +56,11 @@ class DashboardWatcher {
     // Check for changes
     if (uid !== this.uid) {
       this.leave();
-      this.channel = live.getChannel(LiveChannelScope.Grafana, 'dashboard', uid);
+      this.channel = live.getChannel({
+        scope: LiveChannelScope.Grafana,
+        namespace: 'dashboard',
+        path: uid,
+      });
       this.channel.getStream().subscribe(this.observer);
       this.uid = uid;
     }

public/app/features/live/live.ts

@@ -2,7 +2,7 @@ import Centrifuge from 'centrifuge/dist/centrifuge.protobuf';
 import SockJS from 'sockjs-client';
 import { GrafanaLiveSrv, setGrafanaLiveSrv, getGrafanaLiveSrv, config } from '@grafana/runtime';
 import { BehaviorSubject } from 'rxjs';
-import { LiveChannel, LiveChannelScope } from '@grafana/data';
+import { LiveChannel, LiveChannelScope, LiveChannelAddress } from '@grafana/data';
 import { CentrifugeLiveChannel, getErrorChannel } from './channel';
 import {
   GrafanaLiveScope,
@@ -84,23 +84,19 @@ export class CentrifugeSrv implements GrafanaLiveSrv {
    * 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, TPublish>(
-    scopeId: LiveChannelScope,
-    namespace: string,
-    path: string
-  ): LiveChannel<TMessage, TPublish> {
-    const id = `${scopeId}/${namespace}/${path}`;
+  getChannel<TMessage, TPublish = any>(addr: LiveChannelAddress): LiveChannel<TMessage, TPublish> {
+    const id = `${addr.scope}/${addr.namespace}/${addr.path}`;
     let channel = this.open.get(id);
     if (channel != null) {
       return channel;
     }
 
-    const scope = this.scopes[scopeId];
+    const scope = this.scopes[addr.scope];
     if (!scope) {
-      return getErrorChannel('invalid scope', id, scopeId, namespace, path);
+      return getErrorChannel('invalid scope', id, addr);
     }
 
-    channel = new CentrifugeLiveChannel(id, scopeId, namespace, path);
+    channel = new CentrifugeLiveChannel(id, addr);
     channel.shutdownCallback = () => {
       this.open.delete(id); // remove it from the list of open channels
     };
@@ -117,13 +113,14 @@ export class CentrifugeSrv implements GrafanaLiveSrv {
   }
 
   private async initChannel(scope: GrafanaLiveScope, channel: CentrifugeLiveChannel): Promise<void> {
-    const support = await scope.getChannelSupport(channel.namespace);
+    const { addr } = channel;
+    const support = await scope.getChannelSupport(addr.namespace);
     if (!support) {
-      throw new Error(channel.namespace + 'does not support streaming');
+      throw new Error(channel.addr.namespace + 'does not support streaming');
     }
-    const config = support.getChannelConfig(channel.path);
+    const config = support.getChannelConfig(addr.path);
     if (!config) {
-      throw new Error('unknown path: ' + channel.path);
+      throw new Error('unknown path: ' + addr.path);
     }
     if (config.canPublish?.()) {
       channel.publish = (data: any) => this.centrifuge.publish(channel.id, data);

public/app/plugins/datasource/influxdb/components/FluxQueryEditor.tsx

@@ -21,7 +21,7 @@ const samples: Array<SelectableValue<string>> = [
   { label: 'Show buckets', description: 'List the avaliable buckets (table)', value: 'buckets()' },
   {
     label: 'Simple query',
-    description: 'filter by measurment and field',
+    description: 'filter by measurement and field',
     value: `from(bucket: "db/rp")
   |> range(start: v.timeRangeStart, stop:v.timeRangeStop)
   |> filter(fn: (r) =>