From 991f77cee1da71af12d3e853a152537420eb90d7 Mon Sep 17 00:00:00 2001 From: Ryan McKinley Date: Thu, 5 Sep 2019 08:04:01 -0400 Subject: [PATCH] Loki: support loki with streaming in dashboards (#18709) Move some of the buffering with live streaming inside of the datasource, sending full frames instead of deltas and allow Loki in dashboards. --- public/app/core/utils/explore.ts | 2 +- .../dashboard/panel_editor/QueryOptions.tsx | 2 +- .../dashboard/state/PanelQueryRunner.ts | 2 +- .../dashboard/state/PanelQueryState.ts | 2 +- public/app/features/explore/LiveLogs.tsx | 24 +- public/app/features/explore/state/actions.ts | 9 +- public/app/features/explore/state/reducers.ts | 50 +++-- .../explore/utils/ResultProcessor.test.ts | 165 +------------- .../features/explore/utils/ResultProcessor.ts | 32 +-- .../loki/components/LokiQueryEditor.tsx | 132 +++++------ .../loki/components/LokiQueryField.tsx | 2 +- .../datasource/loki/datasource.test.ts | 74 +++++-- .../app/plugins/datasource/loki/datasource.ts | 96 ++++---- .../datasource/loki/live_streams.test.ts | 207 ++++++++++++++++++ .../plugins/datasource/loki/live_streams.ts | 51 +++++ .../app/plugins/datasource/loki/plugin.json | 9 +- .../datasource/loki/result_transformer.ts | 64 +++++- public/app/plugins/datasource/loki/types.ts | 4 + 18 files changed, 538 insertions(+), 389 deletions(-) create mode 100644 public/app/plugins/datasource/loki/live_streams.test.ts create mode 100644 public/app/plugins/datasource/loki/live_streams.ts diff --git a/public/app/core/utils/explore.ts b/public/app/core/utils/explore.ts index 038f283b435..7a862f8a284 100644 --- a/public/app/core/utils/explore.ts +++ b/public/app/core/utils/explore.ts @@ -492,7 +492,7 @@ export enum SortOrder { export const refreshIntervalToSortOrder = (refreshInterval: string) => isLive(refreshInterval) ? SortOrder.Ascending : SortOrder.Descending; -export const sortLogsResult = (logsResult: LogsModel, sortOrder: SortOrder) => { +export const sortLogsResult = (logsResult: LogsModel, sortOrder: SortOrder): LogsModel => { const rows = logsResult ? logsResult.rows : []; sortOrder === SortOrder.Ascending ? rows.sort(sortInAscendingOrder) : rows.sort(sortInDescendingOrder); const result: LogsModel = logsResult ? { ...logsResult, rows } : { hasUniqueLabels: false, rows }; diff --git a/public/app/features/dashboard/panel_editor/QueryOptions.tsx b/public/app/features/dashboard/panel_editor/QueryOptions.tsx index 96773e141ff..512b2857177 100644 --- a/public/app/features/dashboard/panel_editor/QueryOptions.tsx +++ b/public/app/features/dashboard/panel_editor/QueryOptions.tsx @@ -71,7 +71,7 @@ export class QueryOptions extends PureComponent { tooltipInfo: ( <> The maximum data points the query should return. For graphs this is automatically set to one data point per - pixel. + pixel. For some data sources this can also be capped in the datasource settings page. ), }, diff --git a/public/app/features/dashboard/state/PanelQueryRunner.ts b/public/app/features/dashboard/state/PanelQueryRunner.ts index d0576df7279..a8eff9006e3 100644 --- a/public/app/features/dashboard/state/PanelQueryRunner.ts +++ b/public/app/features/dashboard/state/PanelQueryRunner.ts @@ -218,7 +218,7 @@ export class PanelQueryRunner { } /** - * Called after every streaming event. This should be throttled so we + * Called after every streaming event. This should be throttled so we * avoid accidentally overwhelming the browser */ onStreamingDataUpdated = throttle( diff --git a/public/app/features/dashboard/state/PanelQueryState.ts b/public/app/features/dashboard/state/PanelQueryState.ts index 38d4948ad4a..6add79847e8 100644 --- a/public/app/features/dashboard/state/PanelQueryState.ts +++ b/public/app/features/dashboard/state/PanelQueryState.ts @@ -212,7 +212,7 @@ export class PanelQueryState { this.streams = []; - // Move the series from streams to the resposne + // Move the series from streams to the response if (keepSeries) { const { response } = this; this.response = { diff --git a/public/app/features/explore/LiveLogs.tsx b/public/app/features/explore/LiveLogs.tsx index 1d2a6cc9009..9361749426c 100644 --- a/public/app/features/explore/LiveLogs.tsx +++ b/public/app/features/explore/LiveLogs.tsx @@ -1,7 +1,8 @@ import React, { PureComponent } from 'react'; import { css, cx } from 'emotion'; -import { Themeable, withTheme, GrafanaTheme, selectThemeVariant, getLogRowStyles } from '@grafana/ui'; +import { last } from 'lodash'; +import { Themeable, withTheme, GrafanaTheme, selectThemeVariant, getLogRowStyles } from '@grafana/ui'; import { LogsModel, LogRowModel, TimeZone } from '@grafana/data'; import ElapsedTime from './ElapsedTime'; @@ -48,6 +49,7 @@ export interface Props extends Themeable { interface State { logsResultToRender?: LogsModel; + lastTimestamp: number; } class LiveLogs extends PureComponent { @@ -59,6 +61,7 @@ class LiveLogs extends PureComponent { super(props); this.state = { logsResultToRender: props.logsResult, + lastTimestamp: 0, }; } @@ -81,13 +84,17 @@ class LiveLogs extends PureComponent { } } - static getDerivedStateFromProps(nextProps: Props) { + static getDerivedStateFromProps(nextProps: Props, state: State) { if (!nextProps.isPaused) { return { // We update what we show only if not paused. We keep any background subscriptions running and keep updating // our state, but we do not show the updates, this allows us start again showing correct result after resuming // without creating a gap in the log results. logsResultToRender: nextProps.logsResult, + lastTimestamp: + state.logsResultToRender && last(state.logsResultToRender.rows) + ? last(state.logsResultToRender.rows).timeEpochMs + : 0, }; } else { return null; @@ -119,6 +126,15 @@ class LiveLogs extends PureComponent { return rowsToRender; }; + /** + * Check if row is fresh so we can apply special styling. This is bit naive and does not take into account rows + * which arrive out of order. Because loki datasource sends full data instead of deltas we need to compare the + * data and this is easier than doing some intersection of some uuid of each row (which we do not have now anyway) + */ + isFresh = (row: LogRowModel): boolean => { + return row.timeEpochMs > this.state.lastTimestamp; + }; + render() { const { theme, timeZone, onPause, onResume, isPaused } = this.props; const styles = getStyles(theme); @@ -132,10 +148,10 @@ class LiveLogs extends PureComponent { className={cx(['logs-rows', styles.logsRowsLive])} ref={this.scrollContainerRef} > - {this.rowsToRender().map((row: any, index) => { + {this.rowsToRender().map((row: LogRowModel, index) => { return (
{showUtc && ( diff --git a/public/app/features/explore/state/actions.ts b/public/app/features/explore/state/actions.ts index 45942a7e286..5d27bbe18fb 100644 --- a/public/app/features/explore/state/actions.ts +++ b/public/app/features/explore/state/actions.ts @@ -432,6 +432,7 @@ export function runQueries(exploreId: ExploreId): ThunkResult { range, scanning, history, + mode, } = exploreItemState; if (datasourceError) { @@ -454,7 +455,13 @@ export function runQueries(exploreId: ExploreId): ThunkResult { queryState.sendFrames = true; queryState.sendLegacy = true; - const queryOptions = { interval, maxDataPoints: containerWidth, live }; + const queryOptions = { + interval, + // This is used for logs streaming for buffer size. + // TODO: not sure if this makes sense for normal query when using both graph and table + maxDataPoints: mode === ExploreMode.Logs ? 1000 : containerWidth, + live, + }; const datasourceId = datasourceInstance.meta.id; const transaction = buildQueryTransaction(queries, queryOptions, range, queryIntervals, scanning); diff --git a/public/app/features/explore/state/reducers.ts b/public/app/features/explore/state/reducers.ts index f54c0e3dc27..3b6bb0810a7 100644 --- a/public/app/features/explore/state/reducers.ts +++ b/public/app/features/explore/state/reducers.ts @@ -12,7 +12,7 @@ import { } from 'app/core/utils/explore'; import { ExploreItemState, ExploreState, ExploreId, ExploreUpdateState, ExploreMode } from 'app/types/explore'; import { LoadingState } from '@grafana/data'; -import { DataQuery, PanelData } from '@grafana/ui'; +import { DataQuery, DataSourceApi, PanelData } from '@grafana/ui'; import { HigherOrderAction, ActionTypes, @@ -264,25 +264,7 @@ export const itemReducer = reducerFactory({} as ExploreItemSta filter: updateDatasourceInstanceAction, mapper: (state, action): ExploreItemState => { const { datasourceInstance } = action.payload; - // Capabilities - const supportsGraph = datasourceInstance.meta.metrics; - const supportsLogs = datasourceInstance.meta.logs; - - let mode = state.mode || ExploreMode.Metrics; - const supportedModes: ExploreMode[] = []; - - if (supportsGraph) { - supportedModes.push(ExploreMode.Metrics); - } - - if (supportsLogs) { - supportedModes.push(ExploreMode.Logs); - } - - if (supportedModes.length === 1) { - mode = supportedModes[0]; - } - + const [supportedModes, mode] = getModesForDatasource(datasourceInstance, state.mode); // Custom components const StartPage = datasourceInstance.components.ExploreStartPage; stopQueryState(state.queryState, 'Datasource changed'); @@ -586,7 +568,6 @@ export const processQueryResponse = ( ): ExploreItemState => { const { response } = action.payload; const { request, state: loadingState, series, legacy, error } = response; - const replacePreviousResults = action.type === queryEndedAction.type; if (error) { if (error.cancelled) { @@ -615,7 +596,7 @@ export const processQueryResponse = ( } const latency = request.endTime - request.startTime; - const processor = new ResultProcessor(state, replacePreviousResults, series); + const processor = new ResultProcessor(state, series); // For Angular editors state.eventBridge.emit('data-received', legacy); @@ -674,6 +655,31 @@ export const updateChildRefreshState = ( }; }; +const getModesForDatasource = (dataSource: DataSourceApi, currentMode: ExploreMode): [ExploreMode[], ExploreMode] => { + // Temporary hack here. We want Loki to work in dashboards for which it needs to have metrics = true which is weird + // for Explore. + // TODO: need to figure out a better way to handle this situation + const supportsGraph = dataSource.meta.name === 'Loki' ? false : dataSource.meta.metrics; + const supportsLogs = dataSource.meta.logs; + + let mode = currentMode || ExploreMode.Metrics; + const supportedModes: ExploreMode[] = []; + + if (supportsGraph) { + supportedModes.push(ExploreMode.Metrics); + } + + if (supportsLogs) { + supportedModes.push(ExploreMode.Logs); + } + + if (supportedModes.length === 1) { + mode = supportedModes[0]; + } + + return [supportedModes, mode]; +}; + /** * Global Explore reducer that handles multiple Explore areas (left and right). * Actions that have an `exploreId` get routed to the ExploreItemReducer. diff --git a/public/app/features/explore/utils/ResultProcessor.test.ts b/public/app/features/explore/utils/ResultProcessor.test.ts index 75efb92061b..faa649d83ac 100644 --- a/public/app/features/explore/utils/ResultProcessor.test.ts +++ b/public/app/features/explore/utils/ResultProcessor.test.ts @@ -16,7 +16,7 @@ jest.mock('@grafana/data/src/utils/moment_wrapper', () => ({ import { ResultProcessor } from './ResultProcessor'; import { ExploreItemState, ExploreMode } from 'app/types/explore'; import TableModel from 'app/core/table_model'; -import { TimeSeries, LogRowModel, LogsMetaItem, GraphSeriesXY, toDataFrame, FieldType } from '@grafana/data'; +import { TimeSeries, LogRowModel, toDataFrame, FieldType } from '@grafana/data'; const testContext = (options: any = {}) => { const timeSeries = toDataFrame({ @@ -40,7 +40,6 @@ const testContext = (options: any = {}) => { const defaultOptions = { mode: ExploreMode.Metrics, - replacePreviousResults: true, dataFrames: [timeSeries, table], graphResult: [] as TimeSeries[], tableResult: new TableModel(), @@ -57,11 +56,7 @@ const testContext = (options: any = {}) => { queryIntervals: { intervalMs: 10 }, } as any) as ExploreItemState; - const resultProcessor = new ResultProcessor( - state, - combinedOptions.replacePreviousResults, - combinedOptions.dataFrames - ); + const resultProcessor = new ResultProcessor(state, combinedOptions.dataFrames); return { dataFrames: combinedOptions.dataFrames, @@ -206,160 +201,4 @@ describe('ResultProcessor', () => { }); }); }); - - describe('constructed with result that is a DataQueryResponse and merging with previous results', () => { - describe('when calling getLogsResult', () => { - it('then it should return correct logs result', () => { - const { resultProcessor } = testContext({ - mode: ExploreMode.Logs, - replacePreviousResults: false, - logsResult: { - hasUniqueLabels: false, - meta: [], - rows: [ - { - entry: 'This is a previous message 1', - fresh: true, - hasAnsi: false, - labels: { cluster: 'some-cluster' }, - logLevel: 'unknown', - raw: 'This is a previous message 1', - searchWords: [] as string[], - timeEpochMs: 1558038519831, - timeFromNow: 'fromNow() jest mocked', - timeLocal: 'format() jest mocked', - timeUtc: 'format() jest mocked', - timestamp: 1558038519831, - uniqueLabels: {}, - }, - { - entry: 'This is a previous message 2', - fresh: true, - hasAnsi: false, - labels: { cluster: 'some-cluster' }, - logLevel: 'unknown', - raw: 'This is a previous message 2', - searchWords: [] as string[], - timeEpochMs: 1558038518831, - timeFromNow: 'fromNow() jest mocked', - timeLocal: 'format() jest mocked', - timeUtc: 'format() jest mocked', - timestamp: 1558038518831, - uniqueLabels: {}, - }, - ], - series: [ - { - label: 'A-series', - color: '#7EB26D', - data: [[1558038518831, 37.91264531864214], [1558038519831, 38.35179822906545]], - info: undefined, - isVisible: true, - yAxis: { - index: 1, - }, - }, - ], - }, - }); - - const theResult = resultProcessor.getLogsResult(); - const expected = { - hasUniqueLabels: false, - meta: [] as LogsMetaItem[], - rows: [ - { - entry: 'This is a previous message 1', - fresh: false, - hasAnsi: false, - labels: { cluster: 'some-cluster' }, - logLevel: 'unknown', - raw: 'This is a previous message 1', - searchWords: [] as string[], - timeEpochMs: 1558038519831, - timeFromNow: 'fromNow() jest mocked', - timeLocal: 'format() jest mocked', - timeUtc: 'format() jest mocked', - timestamp: 1558038519831, - uniqueLabels: {}, - }, - { - entry: 'This is a previous message 2', - fresh: false, - hasAnsi: false, - labels: { cluster: 'some-cluster' }, - logLevel: 'unknown', - raw: 'This is a previous message 2', - searchWords: [] as string[], - timeEpochMs: 1558038518831, - timeFromNow: 'fromNow() jest mocked', - timeLocal: 'format() jest mocked', - timeUtc: 'format() jest mocked', - timestamp: 1558038518831, - uniqueLabels: {}, - }, - { - entry: 'third', - fresh: true, - hasAnsi: false, - labels: undefined, - logLevel: 'unknown', - raw: 'third', - searchWords: [] as string[], - timeEpochMs: 300, - timeFromNow: 'fromNow() jest mocked', - timeLocal: 'format() jest mocked', - timeUtc: 'format() jest mocked', - timestamp: 300, - uniqueLabels: {}, - }, - { - entry: 'second message', - fresh: true, - hasAnsi: false, - labels: undefined, - logLevel: 'unknown', - raw: 'second message', - searchWords: [] as string[], - timeEpochMs: 200, - timeFromNow: 'fromNow() jest mocked', - timeLocal: 'format() jest mocked', - timeUtc: 'format() jest mocked', - timestamp: 200, - uniqueLabels: {}, - }, - { - entry: 'this is a message', - fresh: true, - hasAnsi: false, - labels: undefined, - logLevel: 'unknown', - raw: 'this is a message', - searchWords: [] as string[], - timeEpochMs: 100, - timeFromNow: 'fromNow() jest mocked', - timeLocal: 'format() jest mocked', - timeUtc: 'format() jest mocked', - timestamp: 100, - uniqueLabels: {}, - }, - ], - series: [ - { - label: 'A-series', - color: '#7EB26D', - data: [[100, 4], [200, 5], [300, 6]], - info: undefined, - isVisible: true, - yAxis: { - index: 1, - }, - } as GraphSeriesXY, - ], - }; - - expect(theResult).toEqual(expected); - }); - }); - }); }); diff --git a/public/app/features/explore/utils/ResultProcessor.ts b/public/app/features/explore/utils/ResultProcessor.ts index 956593b7a75..6633e9238cc 100644 --- a/public/app/features/explore/utils/ResultProcessor.ts +++ b/public/app/features/explore/utils/ResultProcessor.ts @@ -7,11 +7,7 @@ import { dataFrameToLogsModel } from 'app/core/logs_model'; import { getGraphSeriesModel } from 'app/plugins/panel/graph2/getGraphSeriesModel'; export class ResultProcessor { - constructor( - private state: ExploreItemState, - private replacePreviousResults: boolean, - private dataFrames: DataFrame[] - ) {} + constructor(private state: ExploreItemState, private dataFrames: DataFrame[]) {} getGraphResult(): GraphSeriesXY[] { if (this.state.mode !== ExploreMode.Metrics) { @@ -79,30 +75,8 @@ export class ResultProcessor { const sortOrder = refreshIntervalToSortOrder(this.state.refreshInterval); const sortedNewResults = sortLogsResult(newResults, sortOrder); - if (this.replacePreviousResults) { - const slice = 1000; - const rows = sortedNewResults.rows.slice(0, slice); - const series = sortedNewResults.series; - - return { ...sortedNewResults, rows, series }; - } - - const prevLogsResult: LogsModel = this.state.logsResult || { hasUniqueLabels: false, rows: [] }; - const sortedLogResult = sortLogsResult(prevLogsResult, sortOrder); - const rowsInState = sortedLogResult.rows; - - const processedRows = []; - for (const row of rowsInState) { - processedRows.push({ ...row, fresh: false }); - } - for (const row of sortedNewResults.rows) { - processedRows.push({ ...row, fresh: true }); - } - - const slice = -1000; - const rows = processedRows.slice(slice); - const series = sortedNewResults.series.slice(slice); - + const rows = sortedNewResults.rows; + const series = sortedNewResults.series; return { ...sortedNewResults, rows, series }; } } diff --git a/public/app/plugins/datasource/loki/components/LokiQueryEditor.tsx b/public/app/plugins/datasource/loki/components/LokiQueryEditor.tsx index d468fe28ccc..556a012c8e6 100644 --- a/public/app/plugins/datasource/loki/components/LokiQueryEditor.tsx +++ b/public/app/plugins/datasource/loki/components/LokiQueryEditor.tsx @@ -1,90 +1,66 @@ // Libraries -import React, { PureComponent } from 'react'; +import React, { memo } from 'react'; // Types -import { QueryEditorProps } from '@grafana/ui'; +import { AbsoluteTimeRange } from '@grafana/data'; +import { QueryEditorProps, Switch, DataSourceStatus } from '@grafana/ui'; import { LokiDatasource } from '../datasource'; import { LokiQuery } from '../types'; -// import { LokiQueryField } from './LokiQueryField'; +import { LokiQueryField } from './LokiQueryField'; +import { useLokiSyntax } from './useLokiSyntax'; type Props = QueryEditorProps; -// interface State { -// query: LokiQuery; -// } +export const LokiQueryEditor = memo(function LokiQueryEditor(props: Props) { + const { query, panelData, datasource, onChange, onRunQuery } = props; -export class LokiQueryEditor extends PureComponent { - // state: State = { - // query: this.props.query, - // }; - // - // onRunQuery = () => { - // const { query } = this.state; - // - // this.props.onChange(query); - // this.props.onRunQuery(); - // }; - // - // onFieldChange = (query: LokiQuery, override?) => { - // this.setState({ - // query: { - // ...this.state.query, - // expr: query.expr, - // }, - // }); - // }; - // - // onFormatChanged = (option: SelectableValue) => { - // this.props.onChange({ - // ...this.state.query, - // resultFormat: option.value, - // }); - // }; - - render() { - // const { query } = this.state; - // const { datasource } = this.props; - // const formatOptions: SelectableValue[] = [ - // { label: 'Time Series', value: 'time_series' }, - // { label: 'Table', value: 'table' }, - // ]; - // - // query.resultFormat = query.resultFormat || 'time_series'; - // const currentFormat = formatOptions.find(item => item.value === query.resultFormat); - - return ( -
-
-
- Loki is currently not supported as dashboard data source. We are working on it! -
-
- {/* - -
-
-
Format as
-