Explore: Support mixed data sources for supplementary query

.betterer.results

@@ -2278,10 +2278,6 @@ exports[`better eslint`] = {
     "public/app/core/history/richHistoryLocalStorageUtils.ts:5381": [
       [0, 0, 0, "Unexpected any. Specify a different type.", "0"]
     ],
-    "public/app/core/logsModel.ts:5381": [
-      [0, 0, 0, "Do not use any type assertions.", "0"],
-      [0, 0, 0, "Do not use any type assertions.", "1"]
-    ],
     "public/app/core/navigation/GrafanaRoute.test.tsx:5381": [
       [0, 0, 0, "Unexpected any. Specify a different type.", "0"],
       [0, 0, 0, "Unexpected any. Specify a different type.", "1"]

packages/grafana-data/src/types/logs.ts

@@ -91,6 +91,7 @@ export interface LogsModel {
   // visibleRange is time range for histogram created from log results
   visibleRange?: AbsoluteTimeRange;
   queries?: DataQuery[];
+  bucketSize?: number;
 }
 
 export interface LogSearchMatch {
@@ -205,6 +206,35 @@ export enum LogsVolumeType {
   Limited = 'Limited',
 }
 
+/**
+ * Custom meta information required by Logs Volume responses
+ */
+export type LogsVolumeCustomMetaData = {
+  absoluteRange: AbsoluteTimeRange;
+  logsVolumeType: LogsVolumeType;
+  datasourceName: string;
+  sourceQuery: DataQuery;
+};
+
+export const getLogsVolumeAbsoluteRange = (
+  dataFrames: DataFrame[],
+  defaultRange: AbsoluteTimeRange
+): AbsoluteTimeRange => {
+  return dataFrames[0].meta?.custom?.absoluteRange || defaultRange;
+};
+
+export const getLogsVolumeDataSourceInfo = (dataFrames: DataFrame[]): { uid: string; name: string; refId: string } => {
+  return {
+    uid: dataFrames[0]?.meta?.custom?.datasourceUid || '',
+    name: dataFrames[0]?.meta?.custom?.datasourceName || '',
+    refId: dataFrames[0]?.meta?.custom?.sourceQuery?.refId || '',
+  };
+};
+
+export const isLogsVolumeLimited = (dataFrames: DataFrame[]) => {
+  return dataFrames.length && dataFrames[0].meta?.custom?.logsVolumeType === LogsVolumeType.Limited;
+};
+
 /**
  * Data sources that support supplementary queries in Explore.
  * This will enable users to see additional data when running original queries.

public/app/core/logsModel.test.ts

@@ -28,8 +28,8 @@ import {
   getSeriesProperties,
   LIMIT_LABEL,
   logSeriesToLogsModel,
-  queryLogsVolume,
   queryLogsSample,
+  queryLogsVolume,
 } from './logsModel';
 
 const FROM = dateTimeParse('2021-06-17 00:00:00', { timeZone: 'utc' });
@@ -1122,8 +1122,9 @@ describe('logs volume', () => {
     datasource: MockObservableDataSourceApi,
     request: DataQueryRequest<TestDataQuery>;
 
-  function createFrame(labels: object, timestamps: number[], values: number[]) {
+  function createFrame(labels: object, timestamps: number[], values: number[], refId: string) {
     return toDataFrame({
+      refId,
       fields: [
         { name: 'Time', type: FieldType.time, values: timestamps },
         {
@@ -1136,20 +1137,13 @@ describe('logs volume', () => {
     });
   }
 
-  function createExpectedFields(levelName: string) {
-    return [
-      expect.objectContaining({ name: 'Time' }),
-      expect.objectContaining({
-        name: 'Value',
-        config: expect.objectContaining({ displayNameFromDS: levelName }),
-      }),
-    ];
-  }
-
   function setup(datasourceSetup: () => void) {
     datasourceSetup();
     request = {
-      targets: [{ target: 'volume query 1' }, { target: 'volume query 2' }],
+      targets: [
+        { refId: 'A', target: 'volume query 1' },
+        { refId: 'B', target: 'volume query 2' },
+      ],
       scopedVars: {},
     } as unknown as DataQueryRequest<TestDataQuery>;
     volumeProvider = queryLogsVolume(datasource, request, {
@@ -1167,37 +1161,39 @@ describe('logs volume', () => {
 
   function setupMultipleResults() {
     // level=unknown
-    const resultAFrame1 = createFrame({ app: 'app01' }, [100, 200, 300], [5, 5, 5]);
+    const resultAFrame1 = createFrame({ app: 'app01' }, [100, 200, 300], [5, 5, 5], 'A');
     // level=error
-    const resultAFrame2 = createFrame({ app: 'app01', level: 'error' }, [100, 200, 300], [0, 1, 0]);
+    const resultAFrame2 = createFrame({ app: 'app01', level: 'error' }, [100, 200, 300], [0, 1, 0], 'B');
     // level=unknown
-    const resultBFrame1 = createFrame({ app: 'app02' }, [100, 200, 300], [1, 2, 3]);
+    const resultBFrame1 = createFrame({ app: 'app02' }, [100, 200, 300], [1, 2, 3], 'A');
     // level=error
-    const resultBFrame2 = createFrame({ app: 'app02', level: 'error' }, [100, 200, 300], [1, 1, 1]);
+    const resultBFrame2 = createFrame({ app: 'app02', level: 'error' }, [100, 200, 300], [1, 1, 1], 'B');
 
     datasource = new MockObservableDataSourceApi('loki', [
       {
+        state: LoadingState.Loading,
         data: [resultAFrame1, resultAFrame2],
       },
       {
+        state: LoadingState.Done,
         data: [resultBFrame1, resultBFrame2],
       },
     ]);
   }
 
   function setupMultipleResultsStreaming() {
     // level=unknown
-    const resultAFrame1 = createFrame({ app: 'app01' }, [100, 200, 300], [5, 5, 5]);
+    const resultAFrame1 = createFrame({ app: 'app01' }, [100, 200, 300], [5, 5, 5], 'A');
     // level=error
-    const resultAFrame2 = createFrame({ app: 'app01', level: 'error' }, [100, 200, 300], [0, 1, 0]);
+    const resultAFrame2 = createFrame({ app: 'app01', level: 'error' }, [100, 200, 300], [0, 1, 0], 'B');
 
     datasource = new MockObservableDataSourceApi('loki', [
       {
         state: LoadingState.Streaming,
         data: [resultAFrame1],
       },
       {
-        state: LoadingState.Streaming,
+        state: LoadingState.Done,
         data: [resultAFrame1, resultAFrame2],
       },
     ]);
@@ -1220,14 +1216,8 @@ describe('logs volume', () => {
             fields: expect.anything(),
             meta: {
               custom: {
-                targets: [
-                  {
-                    target: 'volume query 1',
-                  },
-                  {
-                    target: 'volume query 2',
-                  },
-                ],
+                sourceQuery: { refId: 'A', target: 'volume query 1' },
+                datasourceName: 'loki',
                 logsVolumeType: LogsVolumeType.FullRange,
                 absoluteRange: {
                   from: FROM.valueOf(),
@@ -1242,7 +1232,7 @@ describe('logs volume', () => {
     });
   });
 
-  it('applies correct meta datya when streaming', async () => {
+  it('applies correct meta data when streaming', async () => {
     setup(setupMultipleResultsStreaming);
 
     await expect(volumeProvider).toEmitValuesWith((received) => {
@@ -1255,14 +1245,8 @@ describe('logs volume', () => {
             fields: expect.anything(),
             meta: {
               custom: {
-                targets: [
-                  {
-                    target: 'volume query 1',
-                  },
-                  {
-                    target: 'volume query 2',
-                  },
-                ],
+                sourceQuery: { refId: 'A', target: 'volume query 1' },
+                datasourceName: 'loki',
                 logsVolumeType: LogsVolumeType.FullRange,
                 absoluteRange: {
                   from: FROM.valueOf(),
@@ -1277,22 +1261,6 @@ describe('logs volume', () => {
     });
   });
 
-  it('aggregates data frames by level', async () => {
-    setup(setupMultipleResults);
-
-    await expect(volumeProvider).toEmitValuesWith((received) => {
-      expect(received).toContainEqual({
-        state: LoadingState.Done,
-        error: undefined,
-        data: expect.arrayContaining([
-          expect.objectContaining({
-            fields: expect.arrayContaining(createExpectedFields('error')),
-          }),
-        ]),
-      });
-    });
-  });
-
   it('returns error', async () => {
     setup(setupErrorResponse);
 

public/app/core/logsModel.ts

@@ -1,4 +1,4 @@
-import { size } from 'lodash';
+import { groupBy, size } from 'lodash';
 import { from, isObservable, Observable } from 'rxjs';
 
 import {
@@ -13,7 +13,6 @@ import {
   dateTimeFormatTimeAgo,
   FieldCache,
   FieldColorModeId,
-  FieldConfig,
   FieldType,
   FieldWithIndex,
   findCommonLabels,
@@ -26,8 +25,8 @@ import {
   LogsMetaItem,
   LogsMetaKind,
   LogsModel,
+  LogsVolumeCustomMetaData,
   LogsVolumeType,
-  MutableDataFrame,
   rangeUtil,
   ScopedVars,
   textUtil,
@@ -222,6 +221,7 @@ export function dataFrameToLogsModel(
         absoluteRange
       );
       logsModel.visibleRange = visibleRange;
+      logsModel.bucketSize = bucketSize;
       logsModel.series = makeDataFramesForLogs(sortedRows, bucketSize);
 
       if (logsModel.meta) {
@@ -602,69 +602,22 @@ function getLogVolumeFieldConfig(level: LogLevel, oneLevelDetected: boolean) {
   };
 }
 
-/**
- * Take multiple data frames, sum up values and group by level.
- * Return a list of data frames, each representing single level.
- */
-export function aggregateRawLogsVolume(
-  rawLogsVolume: DataFrame[],
-  extractLevel: (dataFrame: DataFrame) => LogLevel
-): DataFrame[] {
-  const logsVolumeByLevelMap: Partial<Record<LogLevel, DataFrame[]>> = {};
-
-  rawLogsVolume.forEach((dataFrame) => {
-    const level = extractLevel(dataFrame);
-    if (!logsVolumeByLevelMap[level]) {
-      logsVolumeByLevelMap[level] = [];
+const updateLogsVolumeConfig = (
+  dataFrame: DataFrame,
+  extractLevel: (dataFrame: DataFrame) => LogLevel,
+  oneLevelDetected: boolean
+): DataFrame => {
+  dataFrame.fields = dataFrame.fields.map((field) => {
+    if (field.type === FieldType.number) {
+      field.config = {
+        ...field.config,
+        ...getLogVolumeFieldConfig(extractLevel(dataFrame), oneLevelDetected),
+      };
     }
-    logsVolumeByLevelMap[level]!.push(dataFrame);
-  });
-
-  return Object.keys(logsVolumeByLevelMap).map((level: string) => {
-    return aggregateFields(
-      logsVolumeByLevelMap[level as LogLevel]!,
-      getLogVolumeFieldConfig(level as LogLevel, Object.keys(logsVolumeByLevelMap).length === 1)
-    );
+    return field;
   });
-}
-
-/**
- * Aggregate multiple data frames into a single data frame by adding values.
- * Multiple data frames for the same level are passed here to get a single
- * data frame for a given level. Aggregation by level happens in aggregateRawLogsVolume()
- */
-function aggregateFields(dataFrames: DataFrame[], config: FieldConfig): DataFrame {
-  const aggregatedDataFrame = new MutableDataFrame();
-  if (!dataFrames.length) {
-    return aggregatedDataFrame;
-  }
-
-  const totalLength = dataFrames[0].length;
-  const timeField = new FieldCache(dataFrames[0]).getFirstFieldOfType(FieldType.time);
-
-  if (!timeField) {
-    return aggregatedDataFrame;
-  }
-
-  aggregatedDataFrame.addField({ name: 'Time', type: FieldType.time }, totalLength);
-  aggregatedDataFrame.addField({ name: 'Value', type: FieldType.number, config }, totalLength);
-
-  dataFrames.forEach((dataFrame) => {
-    dataFrame.fields.forEach((field) => {
-      if (field.type === FieldType.number) {
-        for (let pointIndex = 0; pointIndex < totalLength; pointIndex++) {
-          const currentValue = aggregatedDataFrame.get(pointIndex).Value;
-          const valueToAdd = field.values.get(pointIndex);
-          const totalValue =
-            currentValue === null && valueToAdd === null ? null : (currentValue || 0) + (valueToAdd || 0);
-          aggregatedDataFrame.set(pointIndex, { Value: totalValue, Time: timeField.values.get(pointIndex) });
-        }
-      }
-    });
-  });
-
-  return aggregatedDataFrame;
-}
+  return dataFrame;
+};
 
 type LogsVolumeQueryOptions<T extends DataQuery> = {
   extractLevel: (dataFrame: DataFrame) => LogLevel;
@@ -694,7 +647,7 @@ export function queryLogsVolume<TQuery extends DataQuery, TOptions extends DataS
   logsVolumeRequest.hideFromInspector = true;
 
   return new Observable((observer) => {
-    let rawLogsVolume: DataFrame[] = [];
+    let logsVolumeData: DataFrame[] = [];
     observer.next({
       state: LoadingState.Loading,
       error: undefined,
@@ -706,11 +659,6 @@ export function queryLogsVolume<TQuery extends DataQuery, TOptions extends DataS
 
     const subscription = queryObservable.subscribe({
       complete: () => {
-        observer.next({
-          state: LoadingState.Done,
-          error: undefined,
-          data: rawLogsVolume,
-        });
         observer.complete();
       },
       next: (dataQueryResponse: DataQueryResponse) => {
@@ -723,24 +671,34 @@ export function queryLogsVolume<TQuery extends DataQuery, TOptions extends DataS
           });
           observer.error(error);
         } else {
-          const aggregatedLogsVolume = aggregateRawLogsVolume(
-            dataQueryResponse.data.map(toDataFrame),
-            options.extractLevel
-          );
-          if (aggregatedLogsVolume[0]) {
-            aggregatedLogsVolume[0].meta = {
+          const framesByRefId = groupBy(dataQueryResponse.data, 'refId');
+          logsVolumeData = dataQueryResponse.data.map((dataFrame) => {
+            let sourceRefId = dataFrame.refId || '';
+            if (sourceRefId.startsWith('log-volume-')) {
+              sourceRefId = sourceRefId.substr('log-volume-'.length);
+            }
+
+            const logsVolumeCustomMetaData: LogsVolumeCustomMetaData = {
+              logsVolumeType: LogsVolumeType.FullRange,
+              absoluteRange: { from: options.range.from.valueOf(), to: options.range.to.valueOf() },
+              datasourceName: datasource.name,
+              sourceQuery: options.targets.find((dataQuery) => dataQuery.refId === sourceRefId)!,
+            };
+
+            dataFrame.meta = {
+              ...dataFrame.meta,
               custom: {
-                targets: options.targets,
-                logsVolumeType: LogsVolumeType.FullRange,
-                absoluteRange: { from: options.range.from.valueOf(), to: options.range.to.valueOf() },
+                ...dataFrame.meta?.custom,
+                ...logsVolumeCustomMetaData,
               },
             };
-          }
-          rawLogsVolume = aggregatedLogsVolume;
+            return updateLogsVolumeConfig(dataFrame, options.extractLevel, framesByRefId[dataFrame.refId].length === 1);
+          });
+
           observer.next({
-            state: dataQueryResponse.state ?? LoadingState.Streaming,
+            state: dataQueryResponse.state,
             error: undefined,
-            data: rawLogsVolume,
+            data: logsVolumeData,
           });
         }
       },