Alerting: Run queries (#33423)

* alertingqueryrunner first edition

* added so we always set name and uid when changing datasource.

* wip.

* wip

* added support for canceling requests.

* util for getting time ranges for expression queries

* remove logs, store data in state

* added structure for marble testing.

* change so the expression buttons doesnt submit form.

* fixed run button.

* replaced mocks with implementation that will set default query + expression.

* fixed so we set a datasource name for the default expression rule.

* improving expression guard.

* Update public/app/features/alerting/components/AlertingQueryEditor.tsx

Co-authored-by: Hugo Häggmark <hugo.haggmark@grafana.com>

* fixed some nits.

* some refactoring after feedback.

* use grafanthemev2

* rename grafanatheme

* fixing so we convert to correct relative time range.

* added some more tests.

* fixing so duplicating query works.

* added some more tests without marbles.

* small refactoring to share code between runRequest and alerting query runner.

Co-authored-by: Marcus Andersson <marcus.andersson@grafana.com>
Co-authored-by: Hugo Häggmark <hugo.haggmark@grafana.com>

public/app/features/query/state/QueryRunner.ts

@@ -8,8 +8,6 @@ import {
   QueryRunnerOptions,
   QueryRunner as QueryRunnerSrv,
   LoadingState,
-  compareArrayValues,
-  compareDataFrameStructures,
 } from '@grafana/data';
 import { getTemplateSrv } from '@grafana/runtime';
 import { getDatasourceSrv } from 'app/features/plugins/datasource_srv';
@@ -17,6 +15,7 @@ import { cloneDeep } from 'lodash';
 import { from, Observable, ReplaySubject, Unsubscribable } from 'rxjs';
 import { first } from 'rxjs/operators';
 import { getNextRequestId } from './PanelQueryRunner';
+import { setStructureRevision } from './processing/revision';
 import { preProcessPanelData, runRequest } from './runRequest';
 
 export class QueryRunner implements QueryRunnerSrv {
@@ -102,23 +101,7 @@ export class QueryRunner implements QueryRunnerSrv {
           this.subscription = runRequest(ds, request).subscribe({
             next: (data) => {
               const results = preProcessPanelData(data, this.lastResult);
-
-              // Indicate if the structure has changed since the last query
-              let structureRev = 1;
-              if (this.lastResult?.structureRev && this.lastResult.series) {
-                structureRev = this.lastResult.structureRev;
-                const sameStructure = compareArrayValues(
-                  results.series,
-                  this.lastResult.series,
-                  compareDataFrameStructures
-                );
-                if (!sameStructure) {
-                  structureRev++;
-                }
-              }
-              results.structureRev = structureRev;
-              this.lastResult = results;
-
+              this.lastResult = setStructureRevision(results, this.lastResult);
               // Store preprocessed query results for applying overrides later on in the pipeline
               this.subject.next(this.lastResult);
             },

public/app/features/query/state/processing/canceler.ts

@@ -0,0 +1,14 @@
+import { BackendSrv } from 'app/core/services/backend_srv';
+import { MonoTypeOperatorFunction } from 'rxjs';
+import { finalize } from 'rxjs/operators';
+
+export function cancelNetworkRequestsOnUnsubscribe<T>(
+  backendSrv: BackendSrv,
+  requestId: string | undefined
+): MonoTypeOperatorFunction<T> {
+  return finalize(() => {
+    if (requestId) {
+      backendSrv.resolveCancelerIfExists(requestId);
+    }
+  });
+}

public/app/features/query/state/processing/revision.ts

@@ -0,0 +1,16 @@
+import { compareArrayValues, compareDataFrameStructures, PanelData } from '@grafana/data';
+
+export const setStructureRevision = (result: PanelData, lastResult: PanelData | undefined) => {
+  let structureRev = 1;
+
+  if (lastResult?.structureRev && lastResult.series) {
+    structureRev = lastResult.structureRev;
+    const sameStructure = compareArrayValues(result.series, lastResult.series, compareDataFrameStructures);
+    if (!sameStructure) {
+      structureRev++;
+    }
+  }
+
+  result.structureRev = structureRev;
+  return result;
+};

public/app/features/query/state/runRequest.ts

@@ -1,7 +1,7 @@
 // Libraries
 import { from, merge, Observable, of, timer } from 'rxjs';
 import { isString, map as isArray } from 'lodash';
-import { catchError, finalize, map, mapTo, share, takeUntil, tap } from 'rxjs/operators';
+import { catchError, map, mapTo, share, takeUntil, tap } from 'rxjs/operators';
 // Utils & Services
 import { backendSrv } from 'app/core/services/backend_srv';
 // Types
@@ -28,6 +28,7 @@ import {
   ExpressionDatasourceUID,
 } from 'app/features/expressions/ExpressionDatasource';
 import { ExpressionQuery } from 'app/features/expressions/types';
+import { cancelNetworkRequestsOnUnsubscribe } from './processing/canceler';
 
 type MapOfResponsePackets = { [str: string]: DataQueryResponse };
 
@@ -155,7 +156,7 @@ export function runRequest(
     tap(emitDataRequestEvent(datasource)),
     // finalize is triggered when subscriber unsubscribes
     // This makes sure any still running network requests are cancelled
-    finalize(cancelNetworkRequestsOnUnsubscribe(request)),
+    cancelNetworkRequestsOnUnsubscribe(backendSrv, request.requestId),
     // this makes it possible to share this observable in takeUntil
     share()
   );
@@ -166,12 +167,6 @@ export function runRequest(
   return merge(timer(200).pipe(mapTo(state.panelData), takeUntil(dataObservable)), dataObservable);
 }
 
-function cancelNetworkRequestsOnUnsubscribe(req: DataQueryRequest) {
-  return () => {
-    backendSrv.resolveCancelerIfExists(req.requestId);
-  };
-}
-
 export function callQueryMethod(
   datasource: DataSourceApi,
   request: DataQueryRequest,