Loki: Base maxDataPoints limits on query type (#28298)

* Base maxLines and maxDataPoints based on query type

* Allow overriding the limit to higher value

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

@@ -86,7 +86,7 @@ describe('LokiDatasource', () => {
         range,
         range,
       };
       };
 
 
-      const req = ds.createRangeQuery(target, options as any);
+      const req = ds.createRangeQuery(target, options as any, 1000);
       expect(req.start).toBeDefined();
       expect(req.start).toBeDefined();
       expect(req.end).toBeDefined();
       expect(req.end).toBeDefined();
       expect(adjustIntervalSpy).toHaveBeenCalledWith(1000, expect.anything());
       expect(adjustIntervalSpy).toHaveBeenCalledWith(1000, expect.anything());
@@ -101,7 +101,7 @@ describe('LokiDatasource', () => {
         intervalMs: 2000,
         intervalMs: 2000,
       };
       };
 
 
-      const req = ds.createRangeQuery(target, options as any);
+      const req = ds.createRangeQuery(target, options as any, 1000);
       expect(req.start).toBeDefined();
       expect(req.start).toBeDefined();
       expect(req.end).toBeDefined();
       expect(req.end).toBeDefined();
       expect(adjustIntervalSpy).toHaveBeenCalledWith(2000, expect.anything());
       expect(adjustIntervalSpy).toHaveBeenCalledWith(2000, expect.anything());
@@ -109,16 +109,21 @@ describe('LokiDatasource', () => {
   });
   });
 
 
   describe('when querying with limits', () => {
   describe('when querying with limits', () => {
-    const runLimitTest = ({ maxDataPoints, maxLines, expectedLimit, done }: any) => {
+    const runLimitTest = ({
+      maxDataPoints = 123,
+      queryMaxLines,
+      dsMaxLines = 456,
+      expectedLimit,
+      done,
+      expr = '{label="val"}',
+    }: any) => {
       let settings: any = {
       let settings: any = {
         url: 'myloggingurl',
         url: 'myloggingurl',
+        jsonData: {
+          maxLines: dsMaxLines,
+        },
       };
       };
 
 
-      if (Number.isFinite(maxLines!)) {
-        const customData = { ...(settings.jsonData || {}), maxLines: 20 };
-        settings = { ...settings, jsonData: customData };
-      }
-
       const templateSrvMock = ({
       const templateSrvMock = ({
         getAdhocFilters: (): any[] => [],
         getAdhocFilters: (): any[] => [],
         replace: (a: string) => a,
         replace: (a: string) => a,
@@ -126,14 +131,8 @@ describe('LokiDatasource', () => {
 
 
       const ds = new LokiDatasource(settings, templateSrvMock, timeSrvStub as any);
       const ds = new LokiDatasource(settings, templateSrvMock, timeSrvStub as any);
 
 
-      const options = getQueryOptions<LokiQuery>({ targets: [{ expr: 'foo', refId: 'B', maxLines: maxDataPoints }] });
-
-      if (Number.isFinite(maxDataPoints!)) {
-        options.maxDataPoints = maxDataPoints;
-      } else {
-        // By default is 500
-        delete options.maxDataPoints;
-      }
+      const options = getQueryOptions<LokiQuery>({ targets: [{ expr, refId: 'B', maxLines: queryMaxLines }] });
+      options.maxDataPoints = maxDataPoints;
 
 
       observableTester().subscribeAndExpectOnComplete<DataQueryResponse>({
       observableTester().subscribeAndExpectOnComplete<DataQueryResponse>({
         observable: ds.query(options).pipe(take(1)),
         observable: ds.query(options).pipe(take(1)),
@@ -147,33 +146,33 @@ describe('LokiDatasource', () => {
       fetchStream.next(testResponse);
       fetchStream.next(testResponse);
     };
     };
 
 
-    it('should use default max lines when no limit given', done => {
+    it('should use datasource max lines when no limit given and it is log query', done => {
       runLimitTest({
       runLimitTest({
-        expectedLimit: 1000,
+        expectedLimit: 456,
         done,
         done,
       });
       });
     });
     });
 
 
-    it('should use custom max lines if limit is set', done => {
+    it('should use custom max lines from query if set and it is logs query', done => {
       runLimitTest({
       runLimitTest({
-        maxLines: 20,
+        queryMaxLines: 20,
         expectedLimit: 20,
         expectedLimit: 20,
         done,
         done,
       });
       });
     });
     });
 
 
-    it('should use custom maxDataPoints if set in request', () => {
+    it('should use custom max lines from query if set and it is logs query even if it is higher than data source limit', done => {
       runLimitTest({
       runLimitTest({
-        maxDataPoints: 500,
+        queryMaxLines: 500,
         expectedLimit: 500,
         expectedLimit: 500,
+        done,
       });
       });
     });
     });
 
 
-    it('should use datasource maxLimit if maxDataPoints is higher', () => {
+    it('should use maxDataPoints if it is metrics query', () => {
       runLimitTest({
       runLimitTest({
-        maxLines: 20,
-        maxDataPoints: 500,
-        expectedLimit: 20,
+        expr: 'rate({label="val"}[10m])',
+        expectedLimit: 123,
       });
       });
     });
     });
   });
   });
@@ -442,7 +441,7 @@ describe('LokiDatasource', () => {
       // Odd timerange/interval combination that would lead to a float step
       // Odd timerange/interval combination that would lead to a float step
       const options = { range, intervalMs: 2000 };
       const options = { range, intervalMs: 2000 };
 
 
-      expect(Number.isInteger(ds.createRangeQuery(query, options as any).step!)).toBeTruthy();
+      expect(Number.isInteger(ds.createRangeQuery(query, options as any, 1000).step!)).toBeTruthy();
     });
     });
   });
   });
 
 

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

@@ -2,6 +2,7 @@
 import { cloneDeep, isEmpty, map as lodashMap } from 'lodash';
 import { cloneDeep, isEmpty, map as lodashMap } from 'lodash';
 import { merge, Observable, of } from 'rxjs';
 import { merge, Observable, of } from 'rxjs';
 import { catchError, map, switchMap } from 'rxjs/operators';
 import { catchError, map, switchMap } from 'rxjs/operators';
+import Prism from 'prismjs';
 
 
 // Types
 // Types
 import {
 import {
@@ -44,6 +45,7 @@ import { LiveStreams, LokiLiveTarget } from './live_streams';
 import LanguageProvider, { rangeToParams } from './language_provider';
 import LanguageProvider, { rangeToParams } from './language_provider';
 import { serializeParams } from '../../../core/utils/fetch';
 import { serializeParams } from '../../../core/utils/fetch';
 import { RowContextOptions } from '@grafana/ui/src/components/Logs/LogRowContextProvider';
 import { RowContextOptions } from '@grafana/ui/src/components/Logs/LogRowContextProvider';
+import syntax from './syntax';
 
 
 export type RangeQueryOptions = DataQueryRequest<LokiQuery> | AnnotationQueryRequest<LokiQuery>;
 export type RangeQueryOptions = DataQueryRequest<LokiQuery> | AnnotationQueryRequest<LokiQuery>;
 export const DEFAULT_MAX_LINES = 1000;
 export const DEFAULT_MAX_LINES = 1000;
@@ -148,7 +150,7 @@ export class LokiDatasource extends DataSourceApi<LokiQuery, LokiOptions> {
     );
     );
   };
   };
 
 
-  createRangeQuery(target: LokiQuery, options: RangeQueryOptions): LokiRangeQueryRequest {
+  createRangeQuery(target: LokiQuery, options: RangeQueryOptions, limit: number): LokiRangeQueryRequest {
     const query = target.expr;
     const query = target.expr;
     let range: { start?: number; end?: number; step?: number } = {};
     let range: { start?: number; end?: number; step?: number } = {};
     if (options.range) {
     if (options.range) {
@@ -174,7 +176,7 @@ export class LokiDatasource extends DataSourceApi<LokiQuery, LokiOptions> {
       ...DEFAULT_QUERY_PARAMS,
       ...DEFAULT_QUERY_PARAMS,
       ...range,
       ...range,
       query,
       query,
-      limit: Math.min((options as DataQueryRequest<LokiQuery>).maxDataPoints || Infinity, this.maxLines),
+      limit,
     };
     };
   }
   }
 
 
@@ -186,31 +188,22 @@ export class LokiDatasource extends DataSourceApi<LokiQuery, LokiOptions> {
     options: RangeQueryOptions,
     options: RangeQueryOptions,
     responseListLength = 1
     responseListLength = 1
   ): Observable<DataQueryResponse> => {
   ): Observable<DataQueryResponse> => {
-    // target.maxLines value already preprocessed
-    // available cases:
-    // 1) empty input -> mapped to NaN, falls back to dataSource.maxLines limit
-    // 2) input with at least 1 character and that is either incorrect (value in the input field is not a number) or negative
-    //    - mapped to 0, falls back to the limit of 0 lines
-    // 3) default case - correct input, mapped to the value from the input field
-
-    let linesLimit = 0;
-    if (target.maxLines === undefined) {
-      // no target.maxLines, using options.maxDataPoints
-      linesLimit = Math.min((options as DataQueryRequest<LokiQuery>).maxDataPoints || Infinity, this.maxLines);
-    } else {
-      // using target.maxLines
-      if (isNaN(target.maxLines)) {
-        linesLimit = this.maxLines;
-      } else {
-        linesLimit = target.maxLines;
-      }
-    }
+    // For metric query we use maxDataPoints from the request options which should be something like width of the
+    // visualisation in pixels. In case of logs request we either use lines limit defined in the query target or
+    // global limit defined for the data source which ever is lower.
+    let maxDataPoints = isMetricsQuery(target.expr)
+      ? // We fallback to maxLines here because maxDataPoints is defined as possibly undefined. Not sure that can
+        // actually happen both Dashboards and Explore should send some value here. If not maxLines does not make that
+        // much sense but nor any other arbitrary value.
+        (options as DataQueryRequest<LokiQuery>).maxDataPoints || this.maxLines
+      : // If user wants maxLines 0 we still fallback to data source limit. I think that makes sense as why would anyone
+        // want to do a query and not see any results?
+        target.maxLines || this.maxLines;
 
 
-    const queryOptions = { ...options, maxDataPoints: linesLimit };
     if ((options as DataQueryRequest<LokiQuery>).liveStreaming) {
     if ((options as DataQueryRequest<LokiQuery>).liveStreaming) {
-      return this.runLiveQuery(target, queryOptions);
+      return this.runLiveQuery(target, maxDataPoints);
     }
     }
-    const query = this.createRangeQuery(target, queryOptions);
+    const query = this.createRangeQuery(target, options, maxDataPoints);
     return this._request(RANGE_QUERY_ENDPOINT, query).pipe(
     return this._request(RANGE_QUERY_ENDPOINT, query).pipe(
       catchError((err: any) => this.throwUnless(err, err.status === 404, target)),
       catchError((err: any) => this.throwUnless(err, err.status === 404, target)),
       switchMap((response: { data: LokiResponse; status: number }) =>
       switchMap((response: { data: LokiResponse; status: number }) =>
@@ -219,7 +212,7 @@ export class LokiDatasource extends DataSourceApi<LokiQuery, LokiOptions> {
           target,
           target,
           query,
           query,
           responseListLength,
           responseListLength,
-          linesLimit,
+          maxDataPoints,
           this.instanceSettings.jsonData,
           this.instanceSettings.jsonData,
           (options as DataQueryRequest<LokiQuery>).scopedVars,
           (options as DataQueryRequest<LokiQuery>).scopedVars,
           (options as DataQueryRequest<LokiQuery>).reverse
           (options as DataQueryRequest<LokiQuery>).reverse
@@ -228,7 +221,7 @@ export class LokiDatasource extends DataSourceApi<LokiQuery, LokiOptions> {
     );
     );
   };
   };
 
 
-  createLiveTarget(target: LokiQuery, options: { maxDataPoints?: number }): LokiLiveTarget {
+  createLiveTarget(target: LokiQuery, maxDataPoints: number): LokiLiveTarget {
     const query = target.expr;
     const query = target.expr;
     const baseUrl = this.instanceSettings.url;
     const baseUrl = this.instanceSettings.url;
     const params = serializeParams({ query });
     const params = serializeParams({ query });
@@ -237,7 +230,7 @@ export class LokiDatasource extends DataSourceApi<LokiQuery, LokiOptions> {
       query,
       query,
       url: convertToWebSocketUrl(`${baseUrl}/loki/api/v1/tail?${params}`),
       url: convertToWebSocketUrl(`${baseUrl}/loki/api/v1/tail?${params}`),
       refId: target.refId,
       refId: target.refId,
-      size: Math.min(options.maxDataPoints || Infinity, this.maxLines),
+      size: maxDataPoints,
     };
     };
   }
   }
 
 
@@ -247,8 +240,8 @@ export class LokiDatasource extends DataSourceApi<LokiQuery, LokiOptions> {
    * Loki streams, sets only common labels on dataframe.labels and has additional dataframe.fields.labels for unique
    * Loki streams, sets only common labels on dataframe.labels and has additional dataframe.fields.labels for unique
    * labels per row.
    * labels per row.
    */
    */
-  runLiveQuery = (target: LokiQuery, options: { maxDataPoints?: number }): Observable<DataQueryResponse> => {
-    const liveTarget = this.createLiveTarget(target, options);
+  runLiveQuery = (target: LokiQuery, maxDataPoints: number): Observable<DataQueryResponse> => {
+    const liveTarget = this.createLiveTarget(target, maxDataPoints);
 
 
     return this.streams.getStream(liveTarget).pipe(
     return this.streams.getStream(liveTarget).pipe(
       map(data => ({
       map(data => ({
@@ -554,4 +547,16 @@ export function lokiSpecialRegexEscape(value: any) {
   return value;
   return value;
 }
 }
 
 
+/**
+ * Checks if the query expression uses function and so should return a time series instead of logs.
+ * Sometimes important to know that before we actually do the query.
+ */
+function isMetricsQuery(query: string): boolean {
+  const tokens = Prism.tokenize(query, syntax);
+  return tokens.some(t => {
+    // Not sure in which cases it can be string maybe if nothing matched which means it should not be a function
+    return typeof t !== 'string' && t.type === 'function';
+  });
+}
+
 export default LokiDatasource;
 export default LokiDatasource;