Explore: Queries the datasource once per run query and uses DataStreamObserver (#17263)

* Refactor: Removes replaceUrl from actions

* Refactor: Moves saveState thunk to epic

* Refactor: Moves thunks to epics

* Wip: removes resulttype and queries once

* Refactor: LiveTailing uses observer in query

* Refactor: Creates epics folder for epics and move back actioncreators

* Tests: Adds tests for epics and reducer

* Fix: Checks for undefined as well

* Refactor: Cleans up previous live tailing implementation

* Chore: merge with master

* Fix: Fixes url issuses and prom graph in Panels

* Refactor: Removes supportsStreaming and adds sockets to DataSourcePluginMeta instead

* Refactor: Changes the way we create TimeSeries

* Refactor: Renames sockets to streaming

* Refactor: Changes the way Explore does incremental updates

* Refactor: Removes unused method

* Refactor: Adds back Loading indication

public/app/plugins/datasource/loki/datasource.ts

@@ -1,5 +1,8 @@
 // Libraries
 import _ from 'lodash';
+import { Subscription, of } from 'rxjs';
+import { webSocket } from 'rxjs/webSocket';
+import { catchError, map } from 'rxjs/operators';
 
 // Services & Utils
 import * as dateMath from '@grafana/ui/src/utils/datemath';
@@ -17,11 +20,14 @@ import {
   DataSourceInstanceSettings,
   DataQueryError,
   LogRowModel,
+  DataStreamObserver,
+  LoadingState,
+  DataStreamState,
 } from '@grafana/ui';
 import { LokiQuery, LokiOptions } from './types';
 import { BackendSrv } from 'app/core/services/backend_srv';
 import { TemplateSrv } from 'app/features/templating/template_srv';
-import { safeStringifyValue } from 'app/core/utils/explore';
+import { safeStringifyValue, convertToWebSocketUrl } from 'app/core/utils/explore';
 
 export const DEFAULT_MAX_LINES = 1000;
 
@@ -47,6 +53,7 @@ interface LokiContextQueryOptions {
 }
 
 export class LokiDatasource extends DataSourceApi<LokiQuery, LokiOptions> {
+  private subscriptions: { [key: string]: Subscription } = null;
   languageProvider: LanguageProvider;
   maxLines: number;
 
@@ -60,6 +67,7 @@ export class LokiDatasource extends DataSourceApi<LokiQuery, LokiOptions> {
     this.languageProvider = new LanguageProvider(this);
     const settingsData = instanceSettings.jsonData || {};
     this.maxLines = parseInt(settingsData.maxLines, 10) || DEFAULT_MAX_LINES;
+    this.subscriptions = {};
   }
 
   _request(apiUrl: string, data?, options?: any) {
@@ -73,41 +81,20 @@ export class LokiDatasource extends DataSourceApi<LokiQuery, LokiOptions> {
     return this.backendSrv.datasourceRequest(req);
   }
 
-  convertToStreamTargets = (options: DataQueryRequest<LokiQuery>): Array<{ url: string; refId: string }> => {
-    return options.targets
-      .filter(target => target.expr && !target.hide)
-      .map(target => {
-        const interpolated = this.templateSrv.replace(target.expr);
-        const { query, regexp } = parseQuery(interpolated);
-        const refId = target.refId;
-        const baseUrl = this.instanceSettings.url;
-        const params = serializeParams({ query, regexp });
-        const url = `${baseUrl}/api/prom/tail?${params}`;
-
-        return {
-          url,
-          refId,
-        };
-      });
-  };
-
-  resultToSeriesData = (data: any, refId: string): SeriesData[] => {
-    const toSeriesData = (stream: any, refId: string) => ({
-      ...logStreamToSeriesData(stream),
+  prepareLiveTarget(target: LokiQuery, options: DataQueryRequest<LokiQuery>) {
+    const interpolated = this.templateSrv.replace(target.expr);
+    const { query, regexp } = parseQuery(interpolated);
+    const refId = target.refId;
+    const baseUrl = this.instanceSettings.url;
+    const params = serializeParams({ query, regexp });
+    const url = convertToWebSocketUrl(`${baseUrl}/api/prom/tail?${params}`);
+    return {
+      query,
+      regexp,
+      url,
       refId,
-    });
-
-    if (data.streams) {
-      // new Loki API purposed in https://github.com/grafana/loki/pull/590
-      const series: SeriesData[] = [];
-      for (const stream of data.streams || []) {
-        series.push(toSeriesData(stream, refId));
-      }
-      return series;
-    }
-
-    return [toSeriesData(data, refId)];
-  };
+    };
+  }
 
   prepareQueryTarget(target: LokiQuery, options: DataQueryRequest<LokiQuery>) {
     const interpolated = this.templateSrv.replace(target.expr);
@@ -126,9 +113,106 @@ export class LokiDatasource extends DataSourceApi<LokiQuery, LokiOptions> {
     };
   }
 
-  async query(options: DataQueryRequest<LokiQuery>) {
+  unsubscribe = (refId: string) => {
+    const subscription = this.subscriptions[refId];
+    if (subscription && !subscription.closed) {
+      subscription.unsubscribe();
+      delete this.subscriptions[refId];
+    }
+  };
+
+  processError = (err: any, target: any): DataQueryError => {
+    const error: DataQueryError = {
+      message: 'Unknown error during query transaction. Please check JS console logs.',
+      refId: target.refId,
+    };
+
+    if (err.data) {
+      if (typeof err.data === 'string') {
+        error.message = err.data;
+      } else if (err.data.error) {
+        error.message = safeStringifyValue(err.data.error);
+      }
+    } else if (err.message) {
+      error.message = err.message;
+    } else if (typeof err === 'string') {
+      error.message = err;
+    }
+
+    error.status = err.status;
+    error.statusText = err.statusText;
+
+    return error;
+  };
+
+  processResult = (data: any, target: any): SeriesData[] => {
+    const series: SeriesData[] = [];
+
+    if (Object.keys(data).length === 0) {
+      return series;
+    }
+
+    if (!data.streams) {
+      return [{ ...logStreamToSeriesData(data), refId: target.refId }];
+    }
+
+    for (const stream of data.streams || []) {
+      const seriesData = logStreamToSeriesData(stream);
+      seriesData.refId = target.refId;
+      seriesData.meta = {
+        searchWords: getHighlighterExpressionsFromQuery(formatQuery(target.query, target.regexp)),
+        limit: this.maxLines,
+      };
+      series.push(seriesData);
+    }
+
+    return series;
+  };
+
+  runLiveQueries = (options: DataQueryRequest<LokiQuery>, observer?: DataStreamObserver) => {
+    const liveTargets = options.targets
+      .filter(target => target.expr && !target.hide && target.live)
+      .map(target => this.prepareLiveTarget(target, options));
+
+    for (const liveTarget of liveTargets) {
+      const subscription = webSocket(liveTarget.url)
+        .pipe(
+          map((results: any[]) => {
+            const delta = this.processResult(results, liveTarget);
+            const state: DataStreamState = {
+              key: `loki-${liveTarget.refId}`,
+              request: options,
+              state: LoadingState.Streaming,
+              delta,
+              unsubscribe: () => this.unsubscribe(liveTarget.refId),
+            };
+
+            return state;
+          }),
+          catchError(err => {
+            const error = this.processError(err, liveTarget);
+            const state: DataStreamState = {
+              key: `loki-${liveTarget.refId}`,
+              request: options,
+              state: LoadingState.Error,
+              error,
+              unsubscribe: () => this.unsubscribe(liveTarget.refId),
+            };
+
+            return of(state);
+          })
+        )
+        .subscribe({
+          next: state => observer(state),
+        });
+
+      this.subscriptions[liveTarget.refId] = subscription;
+    }
+  };
+
+  runQueries = async (options: DataQueryRequest<LokiQuery>) => {
     const queryTargets = options.targets
-      .filter(target => target.expr && !target.hide)
+      .filter(target => target.expr && !target.hide && !target.live)
       .map(target => this.prepareQueryTarget(target, options));
 
     if (queryTargets.length === 0) {
@@ -141,53 +225,29 @@ export class LokiDatasource extends DataSourceApi<LokiQuery, LokiOptions> {
           return err;
         }
 
-        const error: DataQueryError = {
-          message: 'Unknown error during query transaction. Please check JS console logs.',
-          refId: target.refId,
-        };
-
-        if (err.data) {
-          if (typeof err.data === 'string') {
-            error.message = err.data;
-          } else if (err.data.error) {
-            error.message = safeStringifyValue(err.data.error);
-          }
-        } else if (err.message) {
-          error.message = err.message;
-        } else if (typeof err === 'string') {
-          error.message = err;
-        }
-
-        error.status = err.status;
-        error.statusText = err.statusText;
-
+        const error: DataQueryError = this.processError(err, target);
         throw error;
       })
     );
 
     return Promise.all(queries).then((results: any[]) => {
-      const series: Array<SeriesData | DataQueryError> = [];
+      let series: SeriesData[] = [];
 
       for (let i = 0; i < results.length; i++) {
         const result = results[i];
         if (result.data) {
-          const refId = queryTargets[i].refId;
-          for (const stream of result.data.streams || []) {
-            const seriesData = logStreamToSeriesData(stream);
-            seriesData.refId = refId;
-            seriesData.meta = {
-              searchWords: getHighlighterExpressionsFromQuery(
-                formatQuery(queryTargets[i].query, queryTargets[i].regexp)
-              ),
-              limit: this.maxLines,
-            };
-            series.push(seriesData);
-          }
+          series = series.concat(this.processResult(result.data, queryTargets[i]));
         }
       }
 
       return { data: series };
     });
+  };
+
+  async query(options: DataQueryRequest<LokiQuery>, observer?: DataStreamObserver) {
+    this.runLiveQueries(options, observer);
+
+    return this.runQueries(options);
   }
 
   async importQueries(queries: LokiQuery[], originMeta: PluginMeta): Promise<LokiQuery[]> {

public/app/plugins/datasource/loki/language_provider.ts

@@ -16,6 +16,7 @@ import {
 } from 'app/types/explore';
 import { LokiQuery } from './types';
 import { dateTime } from '@grafana/ui/src/utils/moment_wrapper';
+import { PromQuery } from '../prometheus/types';
 
 const DEFAULT_KEYS = ['job', 'namespace'];
 const EMPTY_SELECTOR = '{}';
@@ -168,8 +169,9 @@ export default class LokiLanguageProvider extends LanguageProvider {
       return Promise.all(
         queries.map(async query => {
           const expr = await this.importPrometheusQuery(query.expr);
+          const { context, ...rest } = query as PromQuery;
           return {
-            ...query,
+            ...rest,
             expr,
           };
         })

public/app/plugins/datasource/loki/plugin.json

@@ -8,6 +8,7 @@
   "alerting": false,
   "annotations": false,
   "logs": true,
+  "streaming": true,
 
   "info": {
     "description": "Like Prometheus but for logs. OSS logging solution from Grafana Labs",

public/app/plugins/datasource/loki/types.ts

@@ -2,6 +2,9 @@ import { DataQuery, Labels, DataSourceJsonData } from '@grafana/ui/src/types';
 
 export interface LokiQuery extends DataQuery {
   expr: string;
+  live?: boolean;
+  query?: string;
+  regexp?: string;
 }
 
 export interface LokiOptions extends DataSourceJsonData {

public/app/plugins/datasource/prometheus/components/PromQueryField.tsx

@@ -223,7 +223,7 @@ class PromQueryField extends React.PureComponent<PromQueryFieldProps, PromQueryF
     // Send text change to parent
     const { query, onChange, onRunQuery } = this.props;
     if (onChange) {
-      const nextQuery: PromQuery = { ...query, expr: value };
+      const nextQuery: PromQuery = { ...query, expr: value, context: 'explore' };
       onChange(nextQuery);
 
       if (override && onRunQuery) {

public/app/plugins/datasource/prometheus/datasource.ts

@@ -1,6 +1,7 @@
 // Libraries
 import _ from 'lodash';
 import $ from 'jquery';
+import { from, Observable } from 'rxjs';
 
 // Services & Utils
 import kbn from 'app/core/utils/kbn';
@@ -14,18 +15,21 @@ import { getQueryHints } from './query_hints';
 import { expandRecordingRules } from './language_utils';
 
 // Types
-import { PromQuery, PromOptions } from './types';
+import { PromQuery, PromOptions, PromQueryRequest } from './types';
 import {
   DataQueryRequest,
   DataSourceApi,
   AnnotationEvent,
   DataSourceInstanceSettings,
   DataQueryError,
+  DataStreamObserver,
+  LoadingState,
 } from '@grafana/ui/src/types';
 import { ExploreUrlState } from 'app/types/explore';
 import { safeStringifyValue } from 'app/core/utils/explore';
 import { TemplateSrv } from 'app/features/templating/template_srv';
 import { TimeSrv } from 'app/features/dashboard/services/TimeSrv';
+import { single, map, filter } from 'rxjs/operators';
 
 export class PrometheusDatasource extends DataSourceApi<PromQuery, PromOptions> {
   type: string;
@@ -83,7 +87,7 @@ export class PrometheusDatasource extends DataSourceApi<PromQuery, PromOptions>
     }
   }
 
-  _request(url, data?, options?: any) {
+  _request(url: string, data?: any, options?: any) {
     options = _.defaults(options || {}, {
       url: this.url + url,
       method: this.httpMethod,
@@ -119,11 +123,11 @@ export class PrometheusDatasource extends DataSourceApi<PromQuery, PromOptions>
   }
 
   // Use this for tab completion features, wont publish response to other components
-  metadataRequest(url) {
+  metadataRequest(url: string) {
     return this._request(url, null, { method: 'GET', silent: true });
   }
 
-  interpolateQueryExpr(value, variable, defaultFormatFn) {
+  interpolateQueryExpr(value: any, variable: any, defaultFormatFn: any) {
     // if no multi or include all do not regexEscape
     if (!variable.multi && !variable.includeAll) {
       return prometheusRegularEscape(value);
@@ -141,34 +145,132 @@ export class PrometheusDatasource extends DataSourceApi<PromQuery, PromOptions>
     return this.templateSrv.variableExists(target.expr);
   }
 
-  query(options: DataQueryRequest<PromQuery>): Promise<{ data: any }> {
-    const start = this.getPrometheusTime(options.range.from, false);
-    const end = this.getPrometheusTime(options.range.to, true);
+  processResult = (response: any, query: PromQueryRequest, target: PromQuery, responseListLength: number) => {
+    // Keeping original start/end for transformers
+    const transformerOptions = {
+      format: target.format,
+      step: query.step,
+      legendFormat: target.legendFormat,
+      start: query.start,
+      end: query.end,
+      query: query.expr,
+      responseListLength,
+      refId: target.refId,
+      valueWithRefId: target.valueWithRefId,
+    };
+    const series = this.resultTransformer.transform(response, transformerOptions);
 
-    const queries = [];
-    const activeTargets = [];
+    return series;
+  };
 
-    options = _.clone(options);
+  runObserverQueries = (
+    options: DataQueryRequest<PromQuery>,
+    observer: DataStreamObserver,
+    queries: PromQueryRequest[],
+    activeTargets: PromQuery[],
+    end: number
+  ) => {
+    for (let index = 0; index < queries.length; index++) {
+      const query = queries[index];
+      const target = activeTargets[index];
+      let observable: Observable<any> = null;
+
+      if (query.instant) {
+        observable = from(this.performInstantQuery(query, end));
+      } else {
+        observable = from(this.performTimeSeriesQuery(query, query.start, query.end));
+      }
+
+      observable
+        .pipe(
+          single(), // unsubscribes automatically after first result
+          filter((response: any) => (response.cancelled ? false : true)),
+          map((response: any) => {
+            return this.processResult(response, query, target, queries.length);
+          })
+        )
+        .subscribe({
+          next: series => {
+            if (query.instant) {
+              observer({
+                key: `prometheus-${target.refId}`,
+                state: LoadingState.Loading,
+                request: options,
+                series: null,
+                delta: series,
+                unsubscribe: () => undefined,
+              });
+            } else {
+              observer({
+                key: `prometheus-${target.refId}`,
+                state: LoadingState.Done,
+                request: options,
+                series: null,
+                delta: series,
+                unsubscribe: () => undefined,
+              });
+            }
+          },
+        });
+    }
+  };
+
+  prepareTargets = (options: DataQueryRequest<PromQuery>, start: number, end: number) => {
+    const queries: PromQueryRequest[] = [];
+    const activeTargets: PromQuery[] = [];
 
     for (const target of options.targets) {
       if (!target.expr || target.hide) {
         continue;
       }
 
+      if (target.context === 'explore') {
+        target.format = 'time_series';
+        target.instant = false;
+        const instantTarget: any = _.cloneDeep(target);
+        instantTarget.format = 'table';
+        instantTarget.instant = true;
+        instantTarget.valueWithRefId = true;
+        delete instantTarget.maxDataPoints;
+        instantTarget.requestId += '_instant';
+        instantTarget.refId += '_instant';
+        activeTargets.push(instantTarget);
+        queries.push(this.createQuery(instantTarget, options, start, end));
+      }
+
       activeTargets.push(target);
       queries.push(this.createQuery(target, options, start, end));
     }
 
+    return {
+      queries,
+      activeTargets,
+    };
+  };
+
+  query(options: DataQueryRequest<PromQuery>, observer?: DataStreamObserver): Promise<{ data: any }> {
+    const start = this.getPrometheusTime(options.range.from, false);
+    const end = this.getPrometheusTime(options.range.to, true);
+
+    options = _.clone(options);
+    const { queries, activeTargets } = this.prepareTargets(options, start, end);
+
     // No valid targets, return the empty result to save a round trip.
     if (_.isEmpty(queries)) {
       return this.$q.when({ data: [] }) as Promise<{ data: any }>;
     }
 
+    if (observer && options.targets.filter(target => target.context === 'explore').length === options.targets.length) {
+      // using observer to make the instant query return immediately
+      this.runObserverQueries(options, observer, queries, activeTargets, end);
+      return this.$q.when({ data: [] }) as Promise<{ data: any }>;
+    }
+
     const allQueryPromise = _.map(queries, query => {
-      if (!query.instant) {
-        return this.performTimeSeriesQuery(query, query.start, query.end);
-      } else {
+      if (query.instant) {
         return this.performInstantQuery(query, end);
+      } else {
+        return this.performTimeSeriesQuery(query, query.start, query.end);
       }
     });
 
@@ -180,19 +282,10 @@ export class PrometheusDatasource extends DataSourceApi<PromQuery, PromOptions>
           return;
         }
 
-        // Keeping original start/end for transformers
-        const transformerOptions = {
-          format: activeTargets[index].format,
-          step: queries[index].step,
-          legendFormat: activeTargets[index].legendFormat,
-          start: queries[index].start,
-          end: queries[index].end,
-          query: queries[index].expr,
-          responseListLength: responseList.length,
-          refId: activeTargets[index].refId,
-          valueWithRefId: activeTargets[index].valueWithRefId,
-        };
-        const series = this.resultTransformer.transform(response, transformerOptions);
+        const target = activeTargets[index];
+        const query = queries[index];
+        const series = this.processResult(response, query, target, queries.length);
+
         result = [...result, ...series];
       });
 
@@ -202,10 +295,16 @@ export class PrometheusDatasource extends DataSourceApi<PromQuery, PromOptions>
     return allPromise as Promise<{ data: any }>;
   }
 
-  createQuery(target, options, start, end) {
-    const query: any = {
+  createQuery(target: PromQuery, options: DataQueryRequest<PromQuery>, start: number, end: number) {
+    const query: PromQueryRequest = {
       hinting: target.hinting,
       instant: target.instant,
+      step: 0,
+      expr: '',
+      requestId: '',
+      refId: '',
+      start: 0,
+      end: 0,
     };
     const range = Math.ceil(end - start);
 
@@ -398,7 +497,7 @@ export class PrometheusDatasource extends DataSourceApi<PromQuery, PromOptions>
     };
     // Unsetting min interval for accurate event resolution
     const minStep = '1s';
-    const query = this.createQuery({ expr, interval: minStep }, queryOptions, start, end);
+    const query = this.createQuery({ expr, interval: minStep, refId: 'X' }, queryOptions, start, end);
 
     const self = this;
     return this.performTimeSeriesQuery(query, query.start, query.end).then(results => {

public/app/plugins/datasource/prometheus/types.ts

@@ -2,6 +2,14 @@ import { DataQuery, DataSourceJsonData } from '@grafana/ui/src/types';
 
 export interface PromQuery extends DataQuery {
   expr: string;
+  context?: 'explore' | 'panel';
+  format?: string;
+  instant?: boolean;
+  hinting?: boolean;
+  interval?: string;
+  intervalFactor?: number;
+  legendFormat?: string;
+  valueWithRefId?: boolean;
 }
 
 export interface PromOptions extends DataSourceJsonData {
@@ -10,3 +18,10 @@ export interface PromOptions extends DataSourceJsonData {
   httpMethod: string;
   directUrl: string;
 }
+
+export interface PromQueryRequest extends PromQuery {
+  step?: number;
+  requestId?: string;
+  start: number;
+  end: number;
+}