From ab2f6fe24cd3c69878be9216f782aee418208907 Mon Sep 17 00:00:00 2001 From: Ryan McKinley Date: Mon, 28 Jun 2021 14:57:10 -0700 Subject: [PATCH] Live: attach stream info to streaming frames (#35465) --- .../src/dataframe/StreamingDataFrame.test.ts | 89 +++++++++++++++++++ .../src/dataframe/StreamingDataFrame.ts | 65 ++++++++++++-- packages/grafana-data/src/dataframe/index.ts | 2 +- packages/grafana-runtime/src/index.ts | 8 +- .../src/utils/DataSourceWithBackend.test.ts | 6 +- .../src/utils/DataSourceWithBackend.ts | 64 +++++++++---- .../state-timeline/StateTimelinePanel.tsx | 8 ++ 7 files changed, 212 insertions(+), 30 deletions(-) diff --git a/packages/grafana-data/src/dataframe/StreamingDataFrame.test.ts b/packages/grafana-data/src/dataframe/StreamingDataFrame.test.ts index bcb608f9aa3..36248bf4268 100644 --- a/packages/grafana-data/src/dataframe/StreamingDataFrame.test.ts +++ b/packages/grafana-data/src/dataframe/StreamingDataFrame.test.ts @@ -376,6 +376,95 @@ describe('Streaming JSON', () => { `); // speed+light 4 ¯\_(ツ)_/¯ better than undefined labels }); + describe('keep track of packets', () => { + const json: DataFrameJSON = { + schema: { + fields: [ + { name: 'time', type: FieldType.time }, + { name: 'value', type: FieldType.number }, + ], + }, + data: { + values: [ + [100, 200, 300], + [1, 2, 3], + ], + }, + }; + + const stream = new StreamingDataFrame(json, { + maxLength: 4, + maxDelta: 300, + }); + + const getSnapshot = (f: StreamingDataFrame) => { + return { + values: f.fields[1].values.toArray(), + info: f.packetInfo, + }; + }; + + expect(getSnapshot(stream)).toMatchInlineSnapshot(` + Object { + "info": Object { + "action": "replace", + "length": 3, + "number": 1, + }, + "values": Array [ + 1, + 2, + 3, + ], + } + `); + + stream.push({ + data: { + values: [ + [400, 500], + [4, 5], + ], + }, + }); + expect(getSnapshot(stream)).toMatchInlineSnapshot(` + Object { + "info": Object { + "action": "append", + "length": 2, + "number": 2, + }, + "values": Array [ + 2, + 3, + 4, + 5, + ], + } + `); + + stream.push({ + data: { + values: [[600], [6]], + }, + }); + expect(getSnapshot(stream)).toMatchInlineSnapshot(` + Object { + "info": Object { + "action": "append", + "length": 1, + "number": 3, + }, + "values": Array [ + 3, + 4, + 5, + 6, + ], + } + `); + }); + /* describe('transpose vertical records', () => { let vrecsA = [ diff --git a/packages/grafana-data/src/dataframe/StreamingDataFrame.ts b/packages/grafana-data/src/dataframe/StreamingDataFrame.ts index 8d18cbeadc8..dbcfe20e916 100644 --- a/packages/grafana-data/src/dataframe/StreamingDataFrame.ts +++ b/packages/grafana-data/src/dataframe/StreamingDataFrame.ts @@ -1,17 +1,40 @@ -import { Field, DataFrame, FieldType } from '../types/dataFrame'; -import { Labels, QueryResultMeta } from '../types'; +import { Field, DataFrame, FieldType, Labels, QueryResultMeta } from '../types'; import { ArrayVector } from '../vector'; import { DataFrameJSON, decodeFieldValueEntities, FieldSchema } from './DataFrameJSON'; import { guessFieldTypeFromValue } from './processDataFrame'; import { join } from '../transformations/transformers/joinDataFrames'; import { AlignedData } from 'uplot'; +/** + * Indicate if the frame is appened or replace + * + * @public -- but runtime + */ +export enum StreamingFrameAction { + Append = 'append', + Replace = 'replace', +} + +/** + * Stream packet info is attached to StreamingDataFrames and indicate how many + * rows were added to the end of the frame. The number of discarded rows can be + * calculated from previous state + * + * @public -- but runtime + */ +export interface StreamPacketInfo { + number: number; + action: StreamingFrameAction; + length: number; +} + /** * @alpha */ export interface StreamingFrameOptions { maxLength?: number; // 1000 maxDelta?: number; // how long to keep things + action?: StreamingFrameAction; // default will append } enum PushMode { @@ -28,7 +51,7 @@ enum PushMode { export class StreamingDataFrame implements DataFrame { name?: string; refId?: string; - meta?: QueryResultMeta; + meta: QueryResultMeta = {}; fields: Array>> = []; length = 0; @@ -38,9 +61,15 @@ export class StreamingDataFrame implements DataFrame { private schemaFields: FieldSchema[] = []; private timeFieldIndex = -1; private pushMode = PushMode.wide; + private alwaysReplace = false; // current labels private labels: Set = new Set(); + readonly packetInfo: StreamPacketInfo = { + number: 0, + action: StreamingFrameAction.Replace, + length: 0, + }; constructor(frame: DataFrameJSON, opts?: StreamingFrameOptions) { this.options = { @@ -48,6 +77,7 @@ export class StreamingDataFrame implements DataFrame { maxDelta: Infinity, ...opts, }; + this.alwaysReplace = this.options.action === StreamingFrameAction.Replace; this.push(frame); } @@ -59,6 +89,8 @@ export class StreamingDataFrame implements DataFrame { push(msg: DataFrameJSON) { const { schema, data } = msg; + this.packetInfo.number++; + if (schema) { this.pushMode = PushMode.wide; this.timeFieldIndex = schema.fields.findIndex((f) => f.type === FieldType.time); @@ -74,7 +106,9 @@ export class StreamingDataFrame implements DataFrame { const niceSchemaFields = this.pushMode === PushMode.labels ? schema.fields.slice(1) : schema.fields; this.refId = schema.refId; - this.meta = schema.meta; + if (schema.meta) { + this.meta = { ...schema.meta }; + } if (hasSameStructure(this.schemaFields, niceSchemaFields)) { const len = niceSchemaFields.length; @@ -163,9 +197,18 @@ export class StreamingDataFrame implements DataFrame { }); } - let curValues = this.fields.map((f) => f.values.buffer); + let appended = values; + this.packetInfo.length = values[0].length; - let appended = circPush(curValues, values, this.options.maxLength, this.timeFieldIndex, this.options.maxDelta); + if (this.alwaysReplace || !this.length) { + this.packetInfo.action = StreamingFrameAction.Replace; + } else { + this.packetInfo.action = StreamingFrameAction.Append; + + // mutates appended + appended = this.fields.map((f) => f.values.buffer); + circPush(appended, values, this.options.maxLength, this.timeFieldIndex, this.options.maxDelta); + } appended.forEach((v, i) => { const { state, values } = this.fields[i]; @@ -265,6 +308,14 @@ function closestIdx(num: number, arr: number[], lo?: number, hi?: number) { return hi; } +/** + * @internal // not exported in yet + */ +export function getLastStreamingDataFramePacket(frame: DataFrame) { + const pi = (frame as StreamingDataFrame).packetInfo; + return pi?.action ? pi : undefined; +} + // mutable circular push function circPush(data: number[][], newData: number[][], maxLength = Infinity, deltaIdx = 0, maxDelta = Infinity) { for (let i = 0; i < data.length; i++) { @@ -296,7 +347,7 @@ function circPush(data: number[][], newData: number[][], maxLength = Infinity, d } } - return data; + return sliceIdx; } function hasSameStructure(a: FieldSchema[], b: FieldSchema[]): boolean { diff --git a/packages/grafana-data/src/dataframe/index.ts b/packages/grafana-data/src/dataframe/index.ts index 75caabf8ef1..1839b8af10f 100644 --- a/packages/grafana-data/src/dataframe/index.ts +++ b/packages/grafana-data/src/dataframe/index.ts @@ -6,6 +6,6 @@ export * from './processDataFrame'; export * from './dimensions'; export * from './ArrayDataFrame'; export * from './DataFrameJSON'; -export { StreamingDataFrame, StreamingFrameOptions } from './StreamingDataFrame'; +export { StreamingDataFrame, StreamingFrameOptions, StreamingFrameAction } from './StreamingDataFrame'; export * from './frameComparisons'; export { anySeriesWithTimeField } from './utils'; diff --git a/packages/grafana-runtime/src/index.ts b/packages/grafana-runtime/src/index.ts index e2fa0650e4f..412af795ede 100644 --- a/packages/grafana-runtime/src/index.ts +++ b/packages/grafana-runtime/src/index.ts @@ -9,7 +9,13 @@ export * from './types'; export { loadPluginCss, SystemJS, PluginCssOptions } from './utils/plugin'; export { reportMetaAnalytics } from './utils/analytics'; export { logInfo, logDebug, logWarning, logError } from './utils/logging'; -export { DataSourceWithBackend, HealthCheckResult, HealthStatus } from './utils/DataSourceWithBackend'; +export { + DataSourceWithBackend, + HealthCheckResult, + HealthCheckResultDetails, + HealthStatus, + StreamOptionsProvider, +} from './utils/DataSourceWithBackend'; export { toDataQueryError, toDataQueryResponse, diff --git a/packages/grafana-runtime/src/utils/DataSourceWithBackend.test.ts b/packages/grafana-runtime/src/utils/DataSourceWithBackend.test.ts index 3ad9e462bf2..e8ec4935a87 100644 --- a/packages/grafana-runtime/src/utils/DataSourceWithBackend.test.ts +++ b/packages/grafana-runtime/src/utils/DataSourceWithBackend.test.ts @@ -1,5 +1,5 @@ import { BackendSrv, BackendSrvRequest } from 'src/services'; -import { DataSourceWithBackend, toStreamingDataResponse } from './DataSourceWithBackend'; +import { DataSourceWithBackend, standardStreamOptionsProvider, toStreamingDataResponse } from './DataSourceWithBackend'; import { DataSourceJsonData, DataQuery, @@ -92,7 +92,7 @@ describe('DataSourceWithBackend', () => { }; // Simple empty query - let obs = toStreamingDataResponse(request, rsp); + let obs = toStreamingDataResponse(rsp, request, standardStreamOptionsProvider); expect(obs).toBeDefined(); let frame = new MutableDataFrame(); @@ -100,7 +100,7 @@ describe('DataSourceWithBackend', () => { channel: 'a/b/c', }; rsp.data = [frame]; - obs = toStreamingDataResponse(request, rsp); + obs = toStreamingDataResponse(rsp, request, standardStreamOptionsProvider); expect(obs).toBeDefined(); }); }); diff --git a/packages/grafana-runtime/src/utils/DataSourceWithBackend.ts b/packages/grafana-runtime/src/utils/DataSourceWithBackend.ts index eb28af7ede5..55f803980ad 100644 --- a/packages/grafana-runtime/src/utils/DataSourceWithBackend.ts +++ b/packages/grafana-runtime/src/utils/DataSourceWithBackend.ts @@ -10,6 +10,7 @@ import { DataFrame, parseLiveChannelAddress, StreamingFrameOptions, + StreamingFrameAction, } from '@grafana/data'; import { merge, Observable, of } from 'rxjs'; import { catchError, switchMap } from 'rxjs/operators'; @@ -140,7 +141,7 @@ class DataSourceWithBackend< const rsp = toDataQueryResponse(raw, queries as DataQuery[]); // Check if any response should subscribe to a live stream if (rsp.data?.length && rsp.data.find((f: DataFrame) => f.meta?.channel)) { - return toStreamingDataResponse(request, rsp); + return toStreamingDataResponse(rsp, request, this.streamOptionsProvider); } return of(rsp); }), @@ -172,6 +173,11 @@ class DataSourceWithBackend< return query; } + /** + * Optionally override the streaming behavior + */ + streamOptionsProvider: StreamOptionsProvider = standardStreamOptionsProvider; + /** * Make a GET request to the datasource resource path */ @@ -218,38 +224,34 @@ class DataSourceWithBackend< } } -export function toStreamingDataResponse( - request: DataQueryRequest, - rsp: DataQueryResponse +/** + * @internal exported for tests + */ +export function toStreamingDataResponse( + rsp: DataQueryResponse, + req: DataQueryRequest, + getter: (req: DataQueryRequest, frame: DataFrame) => StreamingFrameOptions ): Observable { const live = getGrafanaLiveSrv(); if (!live) { return of(rsp); // add warning? } - const buffer: StreamingFrameOptions = { - maxLength: request.maxDataPoints ?? 500, - }; - - // For recent queries, clamp to the current time range - if (request.rangeRaw?.to === 'now') { - buffer.maxDelta = request.range.to.valueOf() - request.range.from.valueOf(); - } - const staticdata: DataFrame[] = []; const streams: Array> = []; - for (const frame of rsp.data) { - const addr = parseLiveChannelAddress(frame.meta?.channel); + for (const f of rsp.data) { + const addr = parseLiveChannelAddress(f.meta?.channel); if (addr) { + const frame = f as DataFrame; streams.push( live.getDataStream({ addr, - buffer, - frame: frame as DataFrame, + buffer: getter(req, frame), + frame, }) ); } else { - staticdata.push(frame); + staticdata.push(f); } } if (staticdata.length) { @@ -261,6 +263,32 @@ export function toStreamingDataResponse( return merge(...streams); } +/** + * This allows data sources to customize the streaming connection query + * + * @public + */ +export type StreamOptionsProvider = ( + request: DataQueryRequest, + frame: DataFrame +) => StreamingFrameOptions; + +/** + * @public + */ +export const standardStreamOptionsProvider: StreamOptionsProvider = (request: DataQueryRequest, frame: DataFrame) => { + const buffer: StreamingFrameOptions = { + maxLength: request.maxDataPoints ?? 500, + action: StreamingFrameAction.Append, + }; + + // For recent queries, clamp to the current time range + if (request.rangeRaw?.to === 'now') { + buffer.maxDelta = request.range.to.valueOf() - request.range.from.valueOf(); + } + return buffer; +}; + //@ts-ignore DataSourceWithBackend = makeClassES5Compatible(DataSourceWithBackend); diff --git a/public/app/plugins/panel/state-timeline/StateTimelinePanel.tsx b/public/app/plugins/panel/state-timeline/StateTimelinePanel.tsx index 21aeecf9c60..06671805735 100755 --- a/public/app/plugins/panel/state-timeline/StateTimelinePanel.tsx +++ b/public/app/plugins/panel/state-timeline/StateTimelinePanel.tsx @@ -5,6 +5,7 @@ import { TimelineMode, TimelineOptions } from './types'; import { TimelineChart } from './TimelineChart'; import { prepareTimelineFields, prepareTimelineLegendItems } from './utils'; import { StateTimelineTooltip } from './StateTimelineTooltip'; +import { getLastStreamingDataFramePacket } from '@grafana/data/src/dataframe/StreamingDataFrame'; interface TimelinePanelProps extends PanelProps {} @@ -61,6 +62,13 @@ export const StateTimelinePanel: React.FC = ({ ); } + if (frames.length === 1) { + const packet = getLastStreamingDataFramePacket(frames[0]); + if (packet) { + // console.log('STREAM Packet', packet); + } + } + return (