Prometheus: Incremental querying option for to: now dashboards

docs/sources/datasources/prometheus/_index.md

@@ -99,6 +99,8 @@ datasources:
       manageAlerts: true
       prometheusType: Prometheus
       prometheusVersion: 2.37.0
+      incrementalQuerying: true
+      incrementalQueryOverlapWindow: 10m
       cacheLevel: 'High'
       exemplarTraceIdDestinations:
         # Field with internal link pointing to data source in Grafana.
@@ -175,3 +177,11 @@ Grafana lists these variables in dropdown select boxes at the top of the dashboa
 Grafana refers to such variables as template variables.
 
 For details, see the [template variables documentation]({{< relref "./template-variables/" >}}).
+
+## Incremental Dashboard Queries (beta)
+
+As of Grafana 10, the Prometheus data source can be configured to query live dashboards incrementally, instead of re-querying the entire duration on each dashboard refresh.
+This can be toggled on or off in the datasource configuration or provisioning file (under `incrementalQuerying` in jsonData).
+Additionally, the amount of overlap between incremental queries can be configured using the `incrementalQueryOverlapWindow` jsonData field, the default value is 10m (10 minutes).
+
+Increasing the duration of the `incrementalQueryOverlapWindow` will increase the size of every incremental query, but might be helpful for instances that have inconsistent results for recent data.

public/app/features/live/data/amendTimeSeries.ts

@@ -0,0 +1,93 @@
+import { closestIdx } from "./StreamingDataFrame";
+
+export type Table = [times: number[], ...values: any[][]];
+
+// prevTable and nextTable are assumed sorted ASC on reference [0] arrays
+// nextTable is assumed to be contiguous, only edges are checked for overlap
+// ...so prev: [1,2,5] + next: [3,4,6] -> [1,2,3,4,6]
+export function amendTable(prevTable: Table, nextTable: Table): Table {
+  let [prevTimes] = prevTable;
+  let [nextTimes] = nextTable;
+
+  let pLen = prevTimes.length;
+  let pStart = prevTimes[0];
+  let pEnd = prevTimes[pLen - 1];
+
+  let nLen = nextTimes.length;
+  let nStart = nextTimes[0];
+  let nEnd = nextTimes[nLen - 1];
+
+  let outTable: Table;
+
+  if (pLen) {
+    if (nLen) {
+      // append, no overlap
+      if (nStart > pEnd) {
+        outTable = prevTable.map((_, i) => prevTable[i].concat(nextTable[i])) as Table;
+      }
+      // prepend, no overlap
+      else if (nEnd < pStart) {
+        outTable = nextTable.map((_, i) => nextTable[i].concat(prevTable[i])) as Table;
+      }
+      // full replace
+      else if (nStart <= pStart && nEnd >= pEnd) {
+        outTable = nextTable;
+      }
+      // partial replace
+      else if (nStart > pStart && nEnd < pEnd) {
+      }
+      // append, with overlap
+      else if (nStart >= pStart) {
+        let idx = closestIdx(nStart, prevTimes);
+        idx = prevTimes[idx] < nStart ? idx - 1 : idx;
+        outTable = prevTable.map((_, i) => prevTable[i].slice(0, idx).concat(nextTable[i])) as Table;
+      }
+      // prepend, with overlap
+      else if (nEnd >= pStart) {
+        let idx = closestIdx(nEnd, prevTimes);
+        idx = prevTimes[idx] < nEnd ? idx : idx + 1;
+        outTable = nextTable.map((_, i) => nextTable[i].concat(prevTable[i].slice(idx))) as Table;
+      }
+    } else {
+      outTable = prevTable;
+    }
+  } else {
+    if (nLen) {
+      outTable = nextTable;
+    } else {
+      outTable = [[]];
+    }
+  }
+
+  return outTable!;
+}
+
+export function trimTable(table: Table, fromTime: number, toTime: number): Table {
+  let [times, ...vals] = table;
+  let fromIdx: number | undefined;
+  let toIdx: number | undefined;
+
+  // trim to bounds
+  if (times[0] < fromTime) {
+    fromIdx = closestIdx(fromTime, times);
+
+    if (times[fromIdx] < fromTime) {
+      fromIdx++;
+    }
+  }
+
+  if (times[times.length - 1] > toTime) {
+    toIdx = closestIdx(toTime, times);
+
+    if (times[toIdx] > toTime) {
+      toIdx--;
+    }
+  }
+
+  if (fromIdx != null || toIdx != null) {
+    times = times.slice(fromIdx ?? 0, toIdx);
+    vals = vals.map(vals2 => vals2.slice(fromIdx ?? 0, toIdx));
+  }
+
+  return [times, ...vals];
+}

public/app/plugins/datasource/prometheus/configuration/PromSettings.tsx

@@ -4,6 +4,7 @@ import semver from 'semver/preload';
 import {
   DataSourcePluginOptionsEditorProps,
   DataSourceSettings as DataSourceSettingsType,
+  isValidDuration,
   onUpdateDatasourceJsonDataOptionChecked,
   SelectableValue,
   updateDatasourcePluginJsonDataOption,
@@ -23,6 +24,7 @@ import config from '../../../../core/config';
 import { useUpdateDatasource } from '../../../../features/datasources/state';
 import { PromApplication, PromBuildInfoResponse } from '../../../../types/unified-alerting-dto';
 import { QueryEditorMode } from '../querybuilder/shared/types';
+import { defaultPrometheusQueryOverlapWindow } from '../querycache/QueryCache';
 import { PrometheusCacheLevel, PromOptions } from '../types';
 
 import { ExemplarsSettings } from './ExemplarsSettings';
@@ -362,6 +364,50 @@ export const PromSettings = (props: Props) => {
             </div>
           </div>
         )}
+
+        <div className="gf-form-inline">
+          <div className="gf-form max-width-30">
+            <FormField
+              label="Incremental querying (beta)"
+              labelWidth={14}
+              tooltip="This feature will change the default behavior of relative queries to always request fresh data from the prometheus instance, instead query results will be cached, and only new records are requested. Turn this on to decrease database and network load."
+              inputEl={
+                <InlineSwitch
+                  value={options.jsonData.incrementalQuerying ?? false}
+                  onChange={onUpdateDatasourceJsonDataOptionChecked(props, 'incrementalQuerying')}
+                  disabled={options.readOnly}
+                />
+              }
+            />
+          </div>
+        </div>
+
+        <div className="gf-form-inline">
+          {options.jsonData.incrementalQuerying && (
+            <FormField
+              label="Query overlap window"
+              labelWidth={14}
+              tooltip="Set a duration like 10m or 120s or 0s. Default of 10 minutes. This duration will be added to the duration of each incremental request."
+              inputEl={
+                <Input
+                  validationEvents={{
+                    onBlur: [
+                      {
+                        rule: (value) => isValidDuration(value),
+                        errorMessage: 'Invalid duration. Example values: 100s, 10m',
+                      },
+                    ],
+                  }}
+                  className="width-25"
+                  value={options.jsonData.incrementalQueryOverlapWindow ?? defaultPrometheusQueryOverlapWindow}
+                  onChange={onChangeHandler('incrementalQueryOverlapWindow', options, onOptionsChange)}
+                  spellCheck={false}
+                  disabled={options.readOnly}
+                />
+              }
+            />
+          )}
+        </div>
       </div>
       <ExemplarsSettings
         options={options.jsonData.exemplarTraceIdDestinations}

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

@@ -56,6 +56,7 @@ import { renderLegendFormat } from './legend';
 import PrometheusMetricFindQuery from './metric_find_query';
 import { getInitHints, getQueryHints } from './query_hints';
 import { QueryEditorMode } from './querybuilder/shared/types';
+import { CacheRequestInfo, defaultPrometheusQueryOverlapWindow, QueryCache } from './querycache/QueryCache';
 import { getOriginalMetricName, transform, transformV2 } from './result_transformer';
 import { trackQuery } from './tracking';
 import {
@@ -84,6 +85,7 @@ export class PrometheusDatasource
 {
   type: string;
   ruleMappings: { [index: string]: string };
+  hasIncrementalQuery: boolean;
   url: string;
   id: number;
   directUrl: string;
@@ -105,6 +107,7 @@ export class PrometheusDatasource
   subType: PromApplication;
   rulerEnabled: boolean;
   cacheLevel: PrometheusCacheLevel;
+  cache: QueryCache;
 
   constructor(
     instanceSettings: DataSourceInstanceSettings<PromOptions>,
@@ -129,6 +132,7 @@ export class PrometheusDatasource
     // here we "fall back" to this.url to make typescript happy, but it should never happen
     this.directUrl = instanceSettings.jsonData.directUrl ?? this.url;
     this.exemplarTraceIdDestinations = instanceSettings.jsonData.exemplarTraceIdDestinations;
+    this.hasIncrementalQuery = instanceSettings.jsonData.incrementalQuerying ?? false;
     this.ruleMappings = {};
     this.languageProvider = languageProvider ?? new PrometheusLanguageProvider(this);
     this.lookupsDisabled = instanceSettings.jsonData.disableMetricsLookup ?? false;
@@ -139,6 +143,9 @@ export class PrometheusDatasource
     this.variables = new PrometheusVariableSupport(this, this.templateSrv, this.timeSrv);
     this.exemplarsAvailable = true;
     this.cacheLevel = instanceSettings.jsonData.cacheLevel ?? PrometheusCacheLevel.Low;
+    this.cache = new QueryCache(
+      instanceSettings.jsonData.incrementalQueryOverlapWindow ?? defaultPrometheusQueryOverlapWindow
+    );
 
     // This needs to be here and cannot be static because of how annotations typing affects casting of data source
     // objects to DataSourceApi types.
@@ -447,12 +454,27 @@ export class PrometheusDatasource
 
   query(request: DataQueryRequest<PromQuery>): Observable<DataQueryResponse> {
     if (this.access === 'proxy') {
-      const targets = request.targets.map((target) => this.processTargetV2(target, request));
+      let fullOrPartialRequest: DataQueryRequest<PromQuery>;
+      let requestInfo: CacheRequestInfo | undefined = undefined;
+      if (this.hasIncrementalQuery) {
+        requestInfo = this.cache.requestInfo(request, this.interpolateString.bind(this));
+        fullOrPartialRequest = requestInfo.requests[0];
+      } else {
+        fullOrPartialRequest = request;
+      }
+
+      const targets = fullOrPartialRequest.targets.map((target) => this.processTargetV2(target, fullOrPartialRequest));
       const startTime = new Date();
-      return super.query({ ...request, targets: targets.flat() }).pipe(
-        map((response) =>
-          transformV2(response, request, { exemplarTraceIdDestinations: this.exemplarTraceIdDestinations })
-        ),
+      return super.query({ ...fullOrPartialRequest, targets: targets.flat() }).pipe(
+        map((response) => {
+          const amendedResponse = {
+            ...response,
+            data: this.cache.procFrames(request, requestInfo, response.data),
+          };
+          return transformV2(amendedResponse, request, {
+            exemplarTraceIdDestinations: this.exemplarTraceIdDestinations,
+          });
+        }),
         tap((response: DataQueryResponse) => {
           trackQuery(response, request, startTime);
         })