From 75527116607e06ce8a9b22dd8cdc6c61054fe0b7 Mon Sep 17 00:00:00 2001 From: Ryan McKinley Date: Mon, 15 Mar 2021 12:16:40 -0700 Subject: [PATCH] Streaming: use StreamingDataFrame for testdata client streaming example (#31998) --- .../src/dataframe/DataFrameJSON.ts | 2 +- .../src/dataframe/StreamingDataFrame.test.ts | 4 +- .../src/dataframe/StreamingDataFrame.ts | 65 +++++++++++-------- .../plugins/datasource/testdata/runStreams.ts | 42 +++++++----- 4 files changed, 66 insertions(+), 47 deletions(-) diff --git a/packages/grafana-data/src/dataframe/DataFrameJSON.ts b/packages/grafana-data/src/dataframe/DataFrameJSON.ts index 6ea5c65a7e5..a2c5785f753 100644 --- a/packages/grafana-data/src/dataframe/DataFrameJSON.ts +++ b/packages/grafana-data/src/dataframe/DataFrameJSON.ts @@ -83,7 +83,7 @@ export interface DataFrameSchema { /** * Field definition without any metadata */ - fields?: FieldSchema[]; + fields: FieldSchema[]; } /** diff --git a/packages/grafana-data/src/dataframe/StreamingDataFrame.test.ts b/packages/grafana-data/src/dataframe/StreamingDataFrame.test.ts index ed281838ab8..af90163c076 100644 --- a/packages/grafana-data/src/dataframe/StreamingDataFrame.test.ts +++ b/packages/grafana-data/src/dataframe/StreamingDataFrame.test.ts @@ -1,5 +1,5 @@ import { FieldType } from '../types/dataFrame'; -import { DataFrameJSON, dataFrameFromJSON } from './DataFrameJSON'; +import { DataFrameJSON } from './DataFrameJSON'; import { StreamingDataFrame } from './StreamingDataFrame'; describe('Streaming JSON', () => { @@ -22,7 +22,7 @@ describe('Streaming JSON', () => { }, }; - const stream = new StreamingDataFrame(dataFrameFromJSON(json)); + const stream = new StreamingDataFrame(json); expect(stream.fields.map((f) => ({ name: f.name, value: f.values.buffer }))).toMatchInlineSnapshot(` Array [ Object { diff --git a/packages/grafana-data/src/dataframe/StreamingDataFrame.ts b/packages/grafana-data/src/dataframe/StreamingDataFrame.ts index 69881052467..57b86830d2f 100644 --- a/packages/grafana-data/src/dataframe/StreamingDataFrame.ts +++ b/packages/grafana-data/src/dataframe/StreamingDataFrame.ts @@ -28,27 +28,12 @@ export class StreamingDataFrame implements DataFrame { private lastUpdateTime = 0; private timeFieldIndex = -1; - constructor(frame: DataFrame, opts?: StreamingFrameOptions) { - this.name = frame.name; - this.refId = frame.refId; - this.meta = frame.meta; + constructor(frame: DataFrameJSON, opts?: StreamingFrameOptions) { this.options = { maxLength: 1000, ...opts, }; - - // Keep the existing fields - this.fields = frame.fields.map((f) => { - if (f.values instanceof ArrayVector) { - return f as Field>; - } - return { - ...f, - values: new ArrayVector(f.values.toArray()), - }; - }); - - this.timeFieldIndex = this.fields.findIndex((f) => f.type === FieldType.time); + this.update(frame); } get length() { @@ -58,14 +43,40 @@ export class StreamingDataFrame implements DataFrame { return this.fields[0].values.length; } + /** + * apply the new message to the existing data. This will replace the existing schema + * if a new schema is included in the message, or append data matching the current schema + */ update(msg: DataFrameJSON) { - if (msg.schema) { - // TODO, replace the existing fields + const { schema, data } = msg; + if (schema) { + if (this.fields.length > 0) { + // ?? keep existing data? + } + + this.name = schema.name; + this.refId = schema.refId; + this.meta = schema.meta; + + // Create new fields from the schema + this.fields = schema.fields.map((f) => { + return { + config: f.config ?? {}, + name: f.name, + labels: f.labels, + type: f.type ?? FieldType.other, + values: new ArrayVector(), + }; + }); + + this.timeFieldIndex = this.fields.findIndex((f) => f.type === FieldType.time); } - if (msg.data) { - const data = msg.data; + if (data && data.values.length && data.values[0].length) { const { values, entities } = data; + if (values.length !== this.fields.length) { + throw new Error('update message mismatch'); + } if (entities) { entities.forEach((ents, i) => { @@ -81,11 +92,14 @@ export class StreamingDataFrame implements DataFrame { }); // Shorten the array less frequently than we append - const elapsed = Date.now() - this.lastUpdateTime; - + const now = Date.now(); + const elapsed = now - this.lastUpdateTime; if (elapsed > 5000) { - if (this.options.maxSeconds && this.timeFieldIndex >= 0) { + if (this.options.maxSeconds && this.timeFieldIndex >= 0 && this.length > 2) { // TODO -- check time length + const tf = this.fields[this.timeFieldIndex].values.buffer; + const elapsed = tf[tf.length - 1] - tf[0]; + console.log('Check elapsed time: ', elapsed); } if (this.options.maxLength) { const delta = this.length - this.options.maxLength; @@ -96,9 +110,8 @@ export class StreamingDataFrame implements DataFrame { }); } } + this.lastUpdateTime = now; } - - this.lastUpdateTime = Date.now(); } } } diff --git a/public/app/plugins/datasource/testdata/runStreams.ts b/public/app/plugins/datasource/testdata/runStreams.ts index d906128c713..2ad9f813873 100644 --- a/public/app/plugins/datasource/testdata/runStreams.ts +++ b/public/app/plugins/datasource/testdata/runStreams.ts @@ -9,6 +9,9 @@ import { CSVReader, Field, LoadingState, + StreamingDataFrame, + DataFrameSchema, + DataFrameData, } from '@grafana/data'; import { TestDataQuery, StreamingQuery } from './types'; @@ -45,32 +48,33 @@ export function runSignalStream( const streamId = `signal-${req.panelId}-${target.refId}`; const maxDataPoints = req.maxDataPoints || 1000; - const data = new CircularDataFrame({ - append: 'tail', - capacity: maxDataPoints, - }); - data.refId = target.refId; - data.name = target.alias || 'Signal ' + target.refId; - data.addField({ name: 'time', type: FieldType.time }); - data.addField({ name: 'value', type: FieldType.number }); + const schema: DataFrameSchema = { + refId: target.refId, + name: target.alias || 'Signal ' + target.refId, + fields: [ + { name: 'time', type: FieldType.time }, + { name: 'value', type: FieldType.number }, + ], + }; const { spread, speed, bands = 0, noise } = query; - for (let i = 0; i < bands; i++) { const suffix = bands > 1 ? ` ${i + 1}` : ''; - data.addField({ name: 'Min' + suffix, type: FieldType.number }); - data.addField({ name: 'Max' + suffix, type: FieldType.number }); + schema.fields.push({ name: 'Min' + suffix, type: FieldType.number }); + schema.fields.push({ name: 'Max' + suffix, type: FieldType.number }); } + const frame = new StreamingDataFrame({ schema }, { maxLength: maxDataPoints }); + let value = Math.random() * 100; let timeoutId: any = null; const addNextRow = (time: number) => { value += (Math.random() - 0.5) * spread; - let idx = 0; - data.fields[idx++].values.add(time); - data.fields[idx++].values.add(value); + const data: DataFrameData = { + values: [[time], [value]], + }; let min = value; let max = value; @@ -79,9 +83,12 @@ export function runSignalStream( min = min - Math.random() * noise; max = max + Math.random() * noise; - data.fields[idx++].values.add(min); - data.fields[idx++].values.add(max); + data.values.push([min]); + data.values.push([max]); } + + const event = { data }; + return frame.update(event); }; // Fill the buffer on init @@ -96,11 +103,10 @@ export function runSignalStream( const pushNextEvent = () => { addNextRow(Date.now()); subscriber.next({ - data: [data], + data: [frame], key: streamId, state: LoadingState.Streaming, }); - timeoutId = setTimeout(pushNextEvent, speed); };