Prometheus: Fix response states (#19092)

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

@@ -0,0 +1,126 @@
+import { PrometheusDatasource } from './datasource';
+import { DataSourceInstanceSettings } from '@grafana/ui';
+import { PromOptions } from './types';
+import { dateTime, LoadingState } from '@grafana/data';
+
+const defaultInstanceSettings: DataSourceInstanceSettings<PromOptions> = {
+  url: 'test_prom',
+  jsonData: {},
+} as any;
+
+const backendSrvMock: any = {
+  datasourceRequest: jest.fn(),
+};
+
+const templateSrvMock: any = {
+  replace(): null {
+    return null;
+  },
+  getAdhocFilters(): any[] {
+    return [];
+  },
+};
+
+const timeSrvMock: any = {
+  timeRange(): any {
+    return {
+      from: dateTime(),
+      to: dateTime(),
+    };
+  },
+};
+
+describe('datasource', () => {
+  describe('query', () => {
+    const ds = new PrometheusDatasource(
+      defaultInstanceSettings,
+      {} as any,
+      backendSrvMock,
+      templateSrvMock,
+      timeSrvMock
+    );
+
+    it('returns empty array when no queries', done => {
+      expect.assertions(2);
+      ds.query(makeQuery([])).subscribe({
+        next(next) {
+          expect(next.data).toEqual([]);
+          expect(next.state).toBe(LoadingState.Done);
+        },
+        complete() {
+          done();
+        },
+      });
+    });
+
+    it('performs time series queries', done => {
+      expect.assertions(2);
+      backendSrvMock.datasourceRequest.mockReturnValueOnce(Promise.resolve(makePromResponse()));
+      ds.query(makeQuery([{}])).subscribe({
+        next(next) {
+          expect(next.data.length).not.toBe(0);
+          expect(next.state).toBe(LoadingState.Done);
+        },
+        complete() {
+          done();
+        },
+      });
+    });
+
+    it('with 2 queries, waits for all to finish until sending Done status', done => {
+      expect.assertions(4);
+      backendSrvMock.datasourceRequest.mockReturnValue(Promise.resolve(makePromResponse()));
+      const responseStatus = [LoadingState.Loading, LoadingState.Done];
+      ds.query(makeQuery([{}, {}])).subscribe({
+        next(next) {
+          expect(next.data.length).not.toBe(0);
+          expect(next.state).toBe(responseStatus.shift());
+        },
+        complete() {
+          done();
+        },
+      });
+    });
+  });
+});
+
+function makeQuery(targets: any[]): any {
+  return {
+    targets: targets.map(t => {
+      return {
+        instant: false,
+        start: dateTime().subtract(5, 'minutes'),
+        end: dateTime(),
+        expr: 'test',
+        ...t,
+      };
+    }),
+    range: {
+      from: dateTime(),
+      to: dateTime(),
+    },
+    interval: '15s',
+  };
+}
+
+/**
+ * Creates a pretty bogus prom response. Definitelly needs more work but right now we do not test the contents of the
+ * messages anyway.
+ */
+function makePromResponse() {
+  return {
+    data: {
+      data: {
+        result: [
+          {
+            metric: {
+              __name__: 'test_metric',
+            },
+            values: [[1568369640, 1]],
+          },
+        ],
+        resultType: 'matrix',
+      },
+    },
+  };
+}

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

@@ -3,9 +3,9 @@ import _ from 'lodash';
 import $ from 'jquery';
 // Services & Utils
 import kbn from 'app/core/utils/kbn';
-import { dateMath, TimeRange, DateTime, AnnotationEvent } from '@grafana/data';
+import { AnnotationEvent, dateMath, DateTime, LoadingState, TimeRange } from '@grafana/data';
-import { Observable, from, of, merge } from 'rxjs';
+import { from, merge, Observable, of } from 'rxjs';
-import { filter, map } from 'rxjs/operators';
+import { filter, map, tap } from 'rxjs/operators';
 
 import PrometheusMetricFindQuery from './metric_find_query';
 import { ResultTransformer } from './result_transformer';
@@ -15,20 +15,19 @@ import addLabelToQuery from './add_label_to_query';
 import { getQueryHints } from './query_hints';
 import { expandRecordingRules } from './language_utils';
 // Types
-import { PromQuery, PromOptions, PromQueryRequest, PromContext } from './types';
+import { PromContext, PromOptions, PromQuery, PromQueryRequest } from './types';
 import {
+  DataQueryError,
   DataQueryRequest,
+  DataQueryResponse,
+  DataQueryResponseData,
   DataSourceApi,
   DataSourceInstanceSettings,
-  DataQueryError,
-  DataQueryResponseData,
-  DataQueryResponse,
 } from '@grafana/ui';
 import { safeStringifyValue } from 'app/core/utils/explore';
 import { TemplateSrv } from 'app/features/templating/template_srv';
 import { TimeSrv } from 'app/features/dashboard/services/TimeSrv';
 import { ExploreUrlState } from 'app/types';
-import { LoadingState } from '@grafana/data/src/types/data';
 
 export interface PromDataQueryResponse {
   data: {
@@ -227,16 +226,16 @@ export class PrometheusDatasource extends DataSourceApi<PromQuery, PromOptions>
 
     // No valid targets, return the empty result to save a round trip.
     if (_.isEmpty(queries)) {
-      return of({ data: [] });
+      return of({
+        data: [],
+        state: LoadingState.Done,
+      });
     }
 
-    const allInstant = queries.filter(query => query.instant).length === queries.length;
+    let runningQueriesCount = queries.length;
-    const allTimeSeries = queries.filter(query => !query.instant).length === queries.length;
     const subQueries = queries.map((query, index) => {
       const target = activeTargets[index];
       let observable: Observable<any> = null;
-      const state: LoadingState =
-        allInstant || allTimeSeries ? LoadingState.Done : query.instant ? LoadingState.Loading : LoadingState.Done;
 
       if (query.instant) {
         observable = from(this.performInstantQuery(query, end));
@@ -245,13 +244,16 @@ export class PrometheusDatasource extends DataSourceApi<PromQuery, PromOptions>
       }
 
       return observable.pipe(
+        // Decrease the counter here. We assume that each request returns only single value and then completes
+        // (should hold until there is some streaming requests involved).
+        tap(() => runningQueriesCount--),
         filter((response: any) => (response.cancelled ? false : true)),
         map((response: any) => {
           const data = this.processResult(response, query, target, queries.length);
           return {
             data,
             key: query.requestId,
-            state,
+            state: runningQueriesCount === 0 ? LoadingState.Done : LoadingState.Loading,
           } as DataQueryResponse;
         })
       );