QueryProcessing: Observable query interface and RxJS for query & stream processing (#18899)
* I needed to learn some rxjs and understand this more, so just playing around * Updated * Removed all the complete calls * Refactoring * StreamHandler -> observable start * progress * simple singal works * Handle update time range * added error handling * wrap old function * minor changes * handle data format in the subscribe function * Use replay subject to return last value to subscribers * Set loading state after no response in 50ms * added missing file * updated comment * Added cancelation of network requests * runRequest: Added unit test scenario framework * Progress on tests * minor refactor of unit tests * updated test * removed some old code * Shared queries work again, and also became so much simplier * unified query and observe methods * implict any fix * Fixed closed subject issue * removed comment * Use last returned data for loading state * WIP: Explore to runRequest makover step1 * Minor progress * Minor progress on explore and runRequest * minor progress * Things are starting to work in explore * Updated prometheus to use new observable query response, greatly simplified code * Revert refId change * Found better solution for key/refId/requestId problem * use observable with loki * tests compile * fix loki query prep * Explore: correct first response handling * Refactorings * Refactoring * Explore: Fixes LoadingState and GraphResults between runs (#18986) * Refactor: Adds state to DataQueryResponse * Fix: Fixes so we do not empty results before new data arrives Fixes: #17409 * Transformations work * observable test data * remove single() from loki promise * Fixed comment * Explore: Fixes failing Loki and Prometheus unit tests (#18995) * Tests: Makes datasource tests work again * Fix: Fixes loki datasource so highligthing works * Chore: Runs Prettier * Fixed query runner tests * Delay loading state indication to 200ms * Fixed test * fixed unit tests * Clear cached calcs * Fixed bug getProcesedDataFrames * Fix the correct test is a better idea * Fix: Fixes so queries in Explore are only run if Graph/Table is shown (#19000) * Fix: Fixes so queries in Explore are only run if Graph/Table is shown Fixes: #18618 * Refactor: Removes unnecessary condition * PanelData: provide legacy data only when needed (#19018) * no legacy * invert logic... now compiles * merge getQueryResponseData and getDataRaw * update comment about query editor * use single getData() function * only send legacy when it is used in explore * pre process rather than post process * pre process rather than post process * Minor refactoring * Add missing tags to test datasource response * MixedDatasource: Adds query observable pattern to MixedDatasource (#19037) * start mixed datasource * Refactor: Refactors into observable parttern * Tests: Fixes tests * Tests: Removes console.log * Refactor: Adds unique requestId
This commit is contained in:
Torkel Ödegaard
+162
@@ -0,0 +1,162 @@
|
||||
import { LogLevel } from '@grafana/data';
|
||||
|
||||
let index = 0;
|
||||
|
||||
export function getRandomLogLevel(): LogLevel {
|
||||
const v = Math.random();
|
||||
if (v > 0.9) {
|
||||
return LogLevel.critical;
|
||||
}
|
||||
if (v > 0.8) {
|
||||
return LogLevel.error;
|
||||
}
|
||||
if (v > 0.7) {
|
||||
return LogLevel.warning;
|
||||
}
|
||||
if (v > 0.4) {
|
||||
return LogLevel.info;
|
||||
}
|
||||
if (v > 0.3) {
|
||||
return LogLevel.debug;
|
||||
}
|
||||
if (v > 0.1) {
|
||||
return LogLevel.trace;
|
||||
}
|
||||
return LogLevel.unknown;
|
||||
}
|
||||
|
||||
export function getNextWord() {
|
||||
index = (index + Math.floor(Math.random() * 5)) % words.length;
|
||||
return words[index];
|
||||
}
|
||||
|
||||
export function getRandomLine(length = 60) {
|
||||
let line = getNextWord();
|
||||
while (line.length < length) {
|
||||
line += ' ' + getNextWord();
|
||||
}
|
||||
return line;
|
||||
}
|
||||
|
||||
const words = [
|
||||
'At',
|
||||
'vero',
|
||||
'eos',
|
||||
'et',
|
||||
'accusamus',
|
||||
'et',
|
||||
'iusto',
|
||||
'odio',
|
||||
'dignissimos',
|
||||
'ducimus',
|
||||
'qui',
|
||||
'blanditiis',
|
||||
'praesentium',
|
||||
'voluptatum',
|
||||
'deleniti',
|
||||
'atque',
|
||||
'corrupti',
|
||||
'quos',
|
||||
'dolores',
|
||||
'et',
|
||||
'quas',
|
||||
'molestias',
|
||||
'excepturi',
|
||||
'sint',
|
||||
'occaecati',
|
||||
'cupiditate',
|
||||
'non',
|
||||
'provident',
|
||||
'similique',
|
||||
'sunt',
|
||||
'in',
|
||||
'culpa',
|
||||
'qui',
|
||||
'officia',
|
||||
'deserunt',
|
||||
'mollitia',
|
||||
'animi',
|
||||
'id',
|
||||
'est',
|
||||
'laborum',
|
||||
'et',
|
||||
'dolorum',
|
||||
'fuga',
|
||||
'Et',
|
||||
'harum',
|
||||
'quidem',
|
||||
'rerum',
|
||||
'facilis',
|
||||
'est',
|
||||
'et',
|
||||
'expedita',
|
||||
'distinctio',
|
||||
'Nam',
|
||||
'libero',
|
||||
'tempore',
|
||||
'cum',
|
||||
'soluta',
|
||||
'nobis',
|
||||
'est',
|
||||
'eligendi',
|
||||
'optio',
|
||||
'cumque',
|
||||
'nihil',
|
||||
'impedit',
|
||||
'quo',
|
||||
'minus',
|
||||
'id',
|
||||
'quod',
|
||||
'maxime',
|
||||
'placeat',
|
||||
'facere',
|
||||
'possimus',
|
||||
'omnis',
|
||||
'voluptas',
|
||||
'assumenda',
|
||||
'est',
|
||||
'omnis',
|
||||
'dolor',
|
||||
'repellendus',
|
||||
'Temporibus',
|
||||
'autem',
|
||||
'quibusdam',
|
||||
'et',
|
||||
'aut',
|
||||
'officiis',
|
||||
'debitis',
|
||||
'aut',
|
||||
'rerum',
|
||||
'necessitatibus',
|
||||
'saepe',
|
||||
'eveniet',
|
||||
'ut',
|
||||
'et',
|
||||
'voluptates',
|
||||
'repudiandae',
|
||||
'sint',
|
||||
'et',
|
||||
'molestiae',
|
||||
'non',
|
||||
'recusandae',
|
||||
'Itaque',
|
||||
'earum',
|
||||
'rerum',
|
||||
'hic',
|
||||
'tenetur',
|
||||
'a',
|
||||
'sapiente',
|
||||
'delectus',
|
||||
'ut',
|
||||
'aut',
|
||||
'reiciendis',
|
||||
'voluptatibus',
|
||||
'maiores',
|
||||
'alias',
|
||||
'consequatur',
|
||||
'aut',
|
||||
'perferendis',
|
||||
'doloribus',
|
||||
'asperiores',
|
||||
'repellat',
|
||||
];
|
||||
+52
-64
@@ -3,95 +3,83 @@ import {
|
||||
DataSourceApi,
|
||||
DataQueryRequest,
|
||||
DataSourceInstanceSettings,
|
||||
DataStreamObserver,
|
||||
DataQueryResponse,
|
||||
MetricFindValue,
|
||||
} from '@grafana/ui';
|
||||
import { TableData, TimeSeries } from '@grafana/data';
|
||||
import { TestDataQuery, Scenario } from './types';
|
||||
import { getBackendSrv } from 'app/core/services/backend_srv';
|
||||
import { StreamHandler } from './StreamHandler';
|
||||
import { queryMetricTree } from './metricTree';
|
||||
import { Observable, from, merge } from 'rxjs';
|
||||
import { runStream } from './runStreams';
|
||||
import templateSrv from 'app/features/templating/template_srv';
|
||||
|
||||
type TestData = TimeSeries | TableData;
|
||||
|
||||
export interface TestDataRegistry {
|
||||
[key: string]: TestData[];
|
||||
}
|
||||
|
||||
export class TestDataDataSource extends DataSourceApi<TestDataQuery> {
|
||||
streams = new StreamHandler();
|
||||
|
||||
/** @ngInject */
|
||||
constructor(instanceSettings: DataSourceInstanceSettings) {
|
||||
super(instanceSettings);
|
||||
}
|
||||
|
||||
query(options: DataQueryRequest<TestDataQuery>, observer: DataStreamObserver) {
|
||||
const queries = options.targets.map(item => {
|
||||
return {
|
||||
...item,
|
||||
intervalMs: options.intervalMs,
|
||||
maxDataPoints: options.maxDataPoints,
|
||||
datasourceId: this.id,
|
||||
alias: templateSrv.replace(item.alias || ''),
|
||||
};
|
||||
});
|
||||
query(options: DataQueryRequest<TestDataQuery>): Observable<DataQueryResponse> {
|
||||
const queries: any[] = [];
|
||||
const streams: Array<Observable<DataQueryResponse>> = [];
|
||||
|
||||
if (queries.length === 0) {
|
||||
return Promise.resolve({ data: [] });
|
||||
// Start streams and prepare queries
|
||||
for (const target of options.targets) {
|
||||
if (target.scenarioId === 'streaming_client') {
|
||||
streams.push(runStream(target, options));
|
||||
} else {
|
||||
queries.push({
|
||||
...target,
|
||||
intervalMs: options.intervalMs,
|
||||
maxDataPoints: options.maxDataPoints,
|
||||
datasourceId: this.id,
|
||||
alias: templateSrv.replace(target.alias || ''),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// Currently we do not support mixed with client only streaming
|
||||
const resp = this.streams.process(options, observer);
|
||||
if (resp) {
|
||||
return Promise.resolve(resp);
|
||||
if (queries.length) {
|
||||
const req: Promise<DataQueryResponse> = getBackendSrv()
|
||||
.datasourceRequest({
|
||||
method: 'POST',
|
||||
url: '/api/tsdb/query',
|
||||
data: {
|
||||
from: options.range.from.valueOf().toString(),
|
||||
to: options.range.to.valueOf().toString(),
|
||||
queries: queries,
|
||||
},
|
||||
// This sets up a cancel token
|
||||
requestId: options.requestId,
|
||||
})
|
||||
.then((res: any) => this.processQueryResult(queries, res));
|
||||
|
||||
streams.push(from(req));
|
||||
}
|
||||
|
||||
return getBackendSrv()
|
||||
.datasourceRequest({
|
||||
method: 'POST',
|
||||
url: '/api/tsdb/query',
|
||||
data: {
|
||||
from: options.range.from.valueOf().toString(),
|
||||
to: options.range.to.valueOf().toString(),
|
||||
queries: queries,
|
||||
},
|
||||
return merge(...streams);
|
||||
}
|
||||
|
||||
// This sets up a cancel token
|
||||
requestId: options.requestId,
|
||||
})
|
||||
.then((res: any) => {
|
||||
const data: TestData[] = [];
|
||||
processQueryResult(queries: any, res: any): DataQueryResponse {
|
||||
const data: TestData[] = [];
|
||||
|
||||
// Returns data in the order it was asked for.
|
||||
// if the response has data with different refId, it is ignored
|
||||
for (const query of queries) {
|
||||
const results = res.data.results[query.refId];
|
||||
if (!results) {
|
||||
console.warn('No Results for:', query);
|
||||
continue;
|
||||
}
|
||||
for (const query of queries) {
|
||||
const results = res.data.results[query.refId];
|
||||
|
||||
for (const t of results.tables || []) {
|
||||
const table = t as TableData;
|
||||
table.refId = query.refId;
|
||||
table.name = query.alias;
|
||||
data.push(table);
|
||||
}
|
||||
for (const t of results.tables || []) {
|
||||
const table = t as TableData;
|
||||
table.refId = query.refId;
|
||||
table.name = query.alias;
|
||||
data.push(table);
|
||||
}
|
||||
|
||||
for (const series of results.series || []) {
|
||||
data.push({
|
||||
target: series.name,
|
||||
datapoints: series.points,
|
||||
refId: query.refId,
|
||||
tags: series.tags,
|
||||
});
|
||||
}
|
||||
}
|
||||
for (const series of results.series || []) {
|
||||
data.push({ target: series.name, datapoints: series.points, refId: query.refId, tags: series.tags });
|
||||
}
|
||||
}
|
||||
|
||||
return { data: data };
|
||||
});
|
||||
return { data };
|
||||
}
|
||||
|
||||
annotationQuery(options: any) {
|
||||
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
import _ from 'lodash';
|
||||
|
||||
import { QueryCtrl } from 'app/plugins/sdk';
|
||||
import { defaultQuery } from './StreamHandler';
|
||||
import { defaultQuery } from './runStreams';
|
||||
import { getBackendSrv } from 'app/core/services/backend_srv';
|
||||
import { dateTime } from '@grafana/data';
|
||||
|
||||
|
||||
@@ -0,0 +1,224 @@
|
||||
import { defaults } from 'lodash';
|
||||
import { Observable } from 'rxjs';
|
||||
|
||||
import { DataQueryRequest, DataQueryResponse } from '@grafana/ui';
|
||||
|
||||
import { FieldType, CircularDataFrame, CSVReader, Field, LoadingState } from '@grafana/data';
|
||||
|
||||
import { TestDataQuery, StreamingQuery } from './types';
|
||||
import { getRandomLine } from './LogIpsum';
|
||||
|
||||
export const defaultQuery: StreamingQuery = {
|
||||
type: 'signal',
|
||||
speed: 250, // ms
|
||||
spread: 3.5,
|
||||
noise: 2.2,
|
||||
bands: 1,
|
||||
};
|
||||
|
||||
export function runStream(target: TestDataQuery, req: DataQueryRequest<TestDataQuery>): Observable<DataQueryResponse> {
|
||||
const query = defaults(target.stream, defaultQuery);
|
||||
if ('signal' === query.type) {
|
||||
return runSignalStream(target, query, req);
|
||||
}
|
||||
if ('logs' === query.type) {
|
||||
return runLogsStream(target, query, req);
|
||||
}
|
||||
if ('fetch' === query.type) {
|
||||
return runFetchStream(target, query, req);
|
||||
}
|
||||
throw new Error(`Unknown Stream Type: ${query.type}`);
|
||||
}
|
||||
|
||||
export function runSignalStream(
|
||||
target: TestDataQuery,
|
||||
query: StreamingQuery,
|
||||
req: DataQueryRequest<TestDataQuery>
|
||||
): Observable<DataQueryResponse> {
|
||||
return new Observable<DataQueryResponse>(subscriber => {
|
||||
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 { spread, speed, bands, 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 });
|
||||
}
|
||||
|
||||
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);
|
||||
|
||||
let min = value;
|
||||
let max = value;
|
||||
|
||||
for (let i = 0; i < bands; i++) {
|
||||
min = min - Math.random() * noise;
|
||||
max = max + Math.random() * noise;
|
||||
|
||||
data.fields[idx++].values.add(min);
|
||||
data.fields[idx++].values.add(max);
|
||||
}
|
||||
};
|
||||
|
||||
// Fill the buffer on init
|
||||
if (true) {
|
||||
let time = Date.now() - maxDataPoints * speed;
|
||||
for (let i = 0; i < maxDataPoints; i++) {
|
||||
addNextRow(time);
|
||||
time += speed;
|
||||
}
|
||||
}
|
||||
|
||||
const pushNextEvent = () => {
|
||||
addNextRow(Date.now());
|
||||
subscriber.next({
|
||||
data: [data],
|
||||
key: streamId,
|
||||
});
|
||||
|
||||
timeoutId = setTimeout(pushNextEvent, speed);
|
||||
};
|
||||
|
||||
// Send first event in 5ms
|
||||
setTimeout(pushNextEvent, 5);
|
||||
|
||||
return () => {
|
||||
console.log('unsubscribing to stream ' + streamId);
|
||||
clearTimeout(timeoutId);
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
export function runLogsStream(
|
||||
target: TestDataQuery,
|
||||
query: StreamingQuery,
|
||||
req: DataQueryRequest<TestDataQuery>
|
||||
): Observable<DataQueryResponse> {
|
||||
return new Observable<DataQueryResponse>(subscriber => {
|
||||
const streamId = `logs-${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 || 'Logs ' + target.refId;
|
||||
data.addField({ name: 'time', type: FieldType.time });
|
||||
data.addField({ name: 'line', type: FieldType.string });
|
||||
|
||||
const { speed } = query;
|
||||
|
||||
let timeoutId: any = null;
|
||||
|
||||
const pushNextEvent = () => {
|
||||
data.values.time.add(Date.now());
|
||||
data.values.line.add(getRandomLine());
|
||||
|
||||
subscriber.next({
|
||||
data: [data],
|
||||
key: streamId,
|
||||
});
|
||||
|
||||
timeoutId = setTimeout(pushNextEvent, speed);
|
||||
};
|
||||
|
||||
// Send first event in 5ms
|
||||
setTimeout(pushNextEvent, 5);
|
||||
|
||||
return () => {
|
||||
console.log('unsubscribing to stream ' + streamId);
|
||||
clearTimeout(timeoutId);
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
export function runFetchStream(
|
||||
target: TestDataQuery,
|
||||
query: StreamingQuery,
|
||||
req: DataQueryRequest<TestDataQuery>
|
||||
): Observable<DataQueryResponse> {
|
||||
return new Observable<DataQueryResponse>(subscriber => {
|
||||
const streamId = `fetch-${req.panelId}-${target.refId}`;
|
||||
const maxDataPoints = req.maxDataPoints || 1000;
|
||||
|
||||
let data = new CircularDataFrame({
|
||||
append: 'tail',
|
||||
capacity: maxDataPoints,
|
||||
});
|
||||
data.refId = target.refId;
|
||||
data.name = target.alias || 'Fetch ' + target.refId;
|
||||
|
||||
let reader: ReadableStreamReader<Uint8Array>;
|
||||
const csv = new CSVReader({
|
||||
callback: {
|
||||
onHeader: (fields: Field[]) => {
|
||||
// Clear any existing fields
|
||||
if (data.fields.length) {
|
||||
data = new CircularDataFrame({
|
||||
append: 'tail',
|
||||
capacity: maxDataPoints,
|
||||
});
|
||||
data.refId = target.refId;
|
||||
data.name = 'Fetch ' + target.refId;
|
||||
}
|
||||
for (const field of fields) {
|
||||
data.addField(field);
|
||||
}
|
||||
},
|
||||
onRow: (row: any[]) => {
|
||||
data.add(row);
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const processChunk = (value: ReadableStreamReadResult<Uint8Array>): any => {
|
||||
if (value.value) {
|
||||
const text = new TextDecoder().decode(value.value);
|
||||
csv.readCSV(text);
|
||||
}
|
||||
|
||||
subscriber.next({
|
||||
data: [data],
|
||||
key: streamId,
|
||||
state: value.done ? LoadingState.Done : LoadingState.Streaming,
|
||||
});
|
||||
|
||||
if (value.done) {
|
||||
console.log('Finished stream');
|
||||
subscriber.complete(); // necessary?
|
||||
return;
|
||||
}
|
||||
|
||||
return reader.read().then(processChunk);
|
||||
};
|
||||
|
||||
fetch(new Request(query.url)).then(response => {
|
||||
reader = response.body.getReader();
|
||||
reader.read().then(processChunk);
|
||||
});
|
||||
|
||||
return () => {
|
||||
// Cancel fetch?
|
||||
console.log('unsubscribing to stream ' + streamId);
|
||||
};
|
||||
});
|
||||
}
|
||||
Reference in New Issue
Block a user