Prometheus: Track incremental query request sizes with Faro (#65244)

Track incremental query usage in Faro

Co-authored-by: leeoniya <leon.sorokin@grafana.com>
Co-authored-by: Galen <galen.kistler@grafana.com>

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

@@ -8,6 +8,8 @@ import {
   isValidDuration,
   parseDuration,
 } from '@grafana/data/src';
+import { faro } from '@grafana/faro-web-sdk';
+import { config } from '@grafana/runtime/src';
 import { amendTable, Table, trimTable } from 'app/features/live/data/amendTimeSeries';
 
 import { PromQuery } from '../types';
@@ -22,6 +24,9 @@ type TargetSig = string;
 
 type TimestampMs = number;
 
+// Look like Q001, Q002, etc
+type RequestID = string;
+
 type StringInterpolator = (expr: string) => string;
 
 // string matching requirements defined in durationutil.ts
@@ -64,8 +69,44 @@ export function getTargSig(targExpr: string, request: DataQueryRequest<PromQuery
  */
 export class QueryCache {
   private overlapWindowMs: number;
+  private perfObeserver?: PerformanceObserver;
+  private shouldProfile: boolean;
+
+  // send profile events every 5 minutes
+  sendEventsInterval = 60000 * 5;
+
+  pendingRequestIdsToTargSigs = new Map<
+    RequestID,
+    {
+      identity: string;
+      bytes: number | null;
+      dashboardUID?: string;
+      interval?: string;
+      panelId?: number;
+      expr?: string;
+    }
+  >();
+
+  pendingAccumulatedEvents = new Map<
+    string,
+    {
+      requestCount: number;
+      savedBytesTotal: number;
+      initialRequestSize: number;
+      lastRequestSize: number;
+      panelId: string;
+      dashId: string;
+      expr: string;
+      interval: string;
+      sent: boolean;
+    }
+  >();
+
+  cache = new Map<TargetIdent, TargetCache>();
+
   constructor(overlapString?: string) {
     const unverifiedOverlap = overlapString ?? defaultPrometheusQueryOverlapWindow;
+
     if (isValidDuration(unverifiedOverlap)) {
       const duration = parseDuration(unverifiedOverlap);
       this.overlapWindowMs = durationToMilliseconds(duration);
@@ -73,14 +114,123 @@ export class QueryCache {
       const duration = parseDuration(defaultPrometheusQueryOverlapWindow);
       this.overlapWindowMs = durationToMilliseconds(duration);
     }
+
+    if (config.grafanaJavascriptAgent.enabled) {
+      this.profile();
+      this.shouldProfile = true;
+    } else {
+      this.shouldProfile = false;
+    }
   }
 
-  cache = new Map<TargetIdent, TargetCache>();
+  private profile() {
+    // Check if PerformanceObserver is supported, and if we have Faro enabled for internal profiling
+    if (typeof PerformanceObserver === 'function') {
+      this.perfObeserver = new PerformanceObserver((list: PerformanceObserverEntryList) => {
+        list.getEntries().forEach((entry) => {
+          // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
+          const entryTypeCast: PerformanceResourceTiming = entry as PerformanceResourceTiming;
+
+          // Safari support for this is coming in 16.4:
+          // https://caniuse.com/mdn-api_performanceresourcetiming_transfersize
+          // Gating that this exists to prevent runtime errors
+          const isSupported = typeof entryTypeCast?.transferSize === 'number';
+
+          if (entryTypeCast?.initiatorType === 'fetch' && isSupported) {
+            let fetchUrl = entryTypeCast.name;
+
+            if (fetchUrl.includes('/api/ds/query')) {
+              let match = fetchUrl.match(/requestId=([a-z\d]+)/i);
+
+              if (match) {
+                let requestId = match[1];
+
+                const requestTransferSize = Math.round(entryTypeCast.transferSize);
+                const currentRequest = this.pendingRequestIdsToTargSigs.get(requestId);
+
+                if (currentRequest) {
+                  const entries = this.pendingRequestIdsToTargSigs.entries();
+
+                  for (let [, value] of entries) {
+                    if (value.identity === currentRequest.identity && value.bytes !== null) {
+                      const previous = this.pendingAccumulatedEvents.get(value.identity);
+
+                      const savedBytes = value.bytes - requestTransferSize;
+
+                      this.pendingAccumulatedEvents.set(value.identity, {
+                        requestCount: (previous?.requestCount ?? 0) + 1,
+                        savedBytesTotal: (previous?.savedBytesTotal ?? 0) + savedBytes,
+                        initialRequestSize: value.bytes,
+                        lastRequestSize: requestTransferSize,
+                        panelId: currentRequest.panelId?.toString() ?? '',
+                        dashId: currentRequest.dashboardUID ?? '',
+                        expr: currentRequest.expr ?? '',
+                        interval: currentRequest.interval ?? '',
+                        sent: false,
+                      });
+
+                      // We don't need to save each subsequent request, only the first one
+                      this.pendingRequestIdsToTargSigs.delete(requestId);
+
+                      return;
+                    }
+                  }
+
+                  // If we didn't return above, this should be the first request, let's save the observed size
+                  this.pendingRequestIdsToTargSigs.set(requestId, { ...currentRequest, bytes: requestTransferSize });
+                }
+              }
+            }
+          }
+        });
+      });
+
+      this.perfObeserver.observe({ type: 'resource', buffered: false });
+
+      setInterval(this.sendPendingTrackingEvents, this.sendEventsInterval);
+
+      // Send any pending profile information when the user navigates away
+      window.addEventListener('beforeunload', this.sendPendingTrackingEvents);
+    }
+  }
+
+  sendPendingTrackingEvents = () => {
+    const entries = this.pendingAccumulatedEvents.entries();
+
+    for (let [key, value] of entries) {
+      if (!value.sent) {
+        this.pendingAccumulatedEvents.set(key, { ...value, sent: true });
+        faro.api.pushMeasurement({
+          type: 'custom',
+          values: {
+            thing: 0,
+            thing2: 1,
+          },
+        });
+        faro.api.pushEvent(
+          'prometheus incremental query response size',
+          {
+            requestCount: value.requestCount.toString(),
+            savedBytesTotal: value.savedBytesTotal.toString(),
+            initialRequestSize: value.initialRequestSize.toString(),
+            lastRequestSize: value.lastRequestSize.toString(),
+            panelId: value.panelId.toString(),
+            dashId: value.dashId.toString(),
+            expr: value.expr.toString(),
+            interval: value.interval.toString(),
+          },
+          'no-interaction',
+          {
+            skipDedupe: true,
+          }
+        );
+      }
+    }
+  };
 
   // can be used to change full range request to partial, split into multiple requests
   requestInfo(request: DataQueryRequest<PromQuery>, interpolateString: StringInterpolator): CacheRequestInfo {
     // TODO: align from/to to interval to increase probability of hitting backend cache
-
     const newFrom = request.range.from.valueOf();
     const newTo = request.range.to.valueOf();
 
@@ -95,9 +245,19 @@ export class QueryCache {
     const reqTargSigs = new Map<TargetIdent, TargetSig>();
     request.targets.forEach((targ) => {
       let targIdent = `${request.dashboardUID}|${request.panelId}|${targ.refId}`;
-      // @todo refactor getTargSig into datasource class and remove targExpr. See #65952 for a potential implementation
       let targExpr = interpolateString(targ.expr);
-      let targSig = getTargSig(targExpr, request, targ);
+      let targSig = getTargSig(targExpr, request, targ); // ${request.maxDataPoints} ?
+
+      if (this.shouldProfile) {
+        this.pendingRequestIdsToTargSigs.set(request.requestId, {
+          identity: targIdent + '|' + targSig,
+          dashboardUID: request.dashboardUID ?? '',
+          interval: targ.interval ?? request.interval,
+          panelId: request.panelId,
+          expr: targExpr,
+          bytes: null,
+        });
+      }
 
       reqTargSigs.set(targIdent, targSig);
     });
@@ -205,12 +365,13 @@ export class QueryCache {
             let nextTable: Table = respFrame.fields.map((field) => field.values.toArray()) as Table;
 
             let amendedTable = amendTable(prevTable, nextTable);
+            if (amendedTable) {
+              for (let i = 0; i < amendedTable.length; i++) {
+                cachedFrame.fields[i].values = new ArrayVector(amendedTable[i]);
+              }
 
-            for (let i = 0; i < amendedTable.length; i++) {
-              cachedFrame.fields[i].values = new ArrayVector(amendedTable[i]);
+              cachedFrame.length = cachedFrame.fields[0].values.length;
             }
-
-            cachedFrame.length = cachedFrame.fields[0].values.length;
           }
         });