Explore: Create DataSourceWithSupplementaryQueriesSupport interface to support log volume and samples (#61298)

* Create DataSourceWithLogsSampleSupport check and move to 1 place

* Add new SupplementaryQueryType

* Add and change utility functions for loading, storing ancd checking supp queries

* Add logic to redux for processing of new type of supp query

* Implement queryLogsSample used to run samples queries

* Fix tests to include also Log samples

* Add tests

* Temporarily, default to false

* Change comment

* Fix lint error

* Refactor handling of supplementary queries in query.ts

* Fix looping over array

* Remove changes for any => unknowns as in utils.ts

* Fix logic

* Fix incorrect imports after function was moved to different file

* Migrate old log volume key

* Update public/app/features/explore/utils/supplementaryQueries.ts

Co-authored-by: Piotr Jamróz <pm.jamroz@gmail.com>

* Refactor to use DataSourceWithSupplementaryQueriesSupport

* Refactor, improve tests,  change internal API

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

Co-authored-by: Piotr Jamróz <pm.jamroz@gmail.com>

* Add deprecation for DataSourceWithLogsVolumeSupport, but still support it

* Update comment with correct new issue

Co-authored-by: Piotr Jamróz <pm.jamroz@gmail.com>

public/app/core/logsModel.test.ts

@@ -28,6 +28,7 @@ import {
   LIMIT_LABEL,
   logSeriesToLogsModel,
   queryLogsVolume,
+  queryLogsSample,
 } from './logsModel';
 
 describe('dedupLogRows()', () => {
@@ -1223,3 +1224,95 @@ describe('logs volume', () => {
     });
   });
 });
+
+describe('logs sample', () => {
+  class TestDataQuery implements DataQuery {
+    refId = 'A';
+    target = '';
+  }
+
+  let logsSampleProvider: Observable<DataQueryResponse>,
+    datasource: MockObservableDataSourceApi,
+    request: DataQueryRequest<TestDataQuery>;
+
+  function createFrame(labels: object[], timestamps: number[], values: string[]) {
+    return toDataFrame({
+      fields: [
+        { name: 'Time', type: FieldType.time, values: timestamps },
+        {
+          name: 'Line',
+          type: FieldType.string,
+          values,
+        },
+        { name: 'labels', type: FieldType.other, values: labels },
+      ],
+    });
+  }
+
+  function setup(datasourceSetup: () => void) {
+    datasourceSetup();
+    request = {
+      targets: [{ target: 'logs sample query 1' }, { target: 'logs sample query 2' }],
+      scopedVars: {},
+    } as unknown as DataQueryRequest<TestDataQuery>;
+    logsSampleProvider = queryLogsSample(datasource, request);
+  }
+  const resultAFrame1 = createFrame([{ app: 'app01' }], [100, 200, 300], ['line 1', 'line 2', 'line 3']);
+  const resultAFrame2 = createFrame(
+    [{ app: 'app01', level: 'error' }],
+    [100, 200, 300],
+    ['line 4', 'line 5', 'line 6']
+  );
+
+  const resultBFrame1 = createFrame([{ app: 'app02' }], [100, 200, 300], ['line A', 'line B', 'line C']);
+  const resultBFrame2 = createFrame(
+    [{ app: 'app02', level: 'error' }],
+    [100, 200, 300],
+    ['line D', 'line E', 'line F']
+  );
+
+  function setupMultipleResults() {
+    datasource = new MockObservableDataSourceApi('loki', [
+      {
+        data: [resultAFrame1, resultAFrame2],
+      },
+      {
+        data: [resultBFrame1, resultBFrame2],
+      },
+    ]);
+  }
+
+  function setupErrorResponse() {
+    datasource = new MockObservableDataSourceApi('loki', [], undefined, 'Error message');
+  }
+
+  it('returns data', async () => {
+    setup(setupMultipleResults);
+    await expect(logsSampleProvider).toEmitValuesWith((received) => {
+      expect(received).toMatchObject([
+        { state: LoadingState.Loading, error: undefined, data: [] },
+        {
+          state: LoadingState.Done,
+          error: undefined,
+          data: [resultAFrame1, resultAFrame2, resultBFrame1, resultBFrame2],
+        },
+      ]);
+    });
+  });
+
+  it('returns error', async () => {
+    setup(setupErrorResponse);
+
+    await expect(logsSampleProvider).toEmitValuesWith((received) => {
+      expect(received).toMatchObject([
+        { state: LoadingState.Loading, error: undefined, data: [] },
+        {
+          state: LoadingState.Error,
+          error: 'Error message',
+          data: [],
+        },
+        'Error message',
+      ]);
+    });
+  });
+});

public/app/core/logsModel.ts

@@ -747,6 +747,63 @@ export function queryLogsVolume<TQuery extends DataQuery, TOptions extends DataS
   });
 }
 
+/**
+ * Creates an observable, which makes requests to get logs samples.
+ */
+export function queryLogsSample<TQuery extends DataQuery, TOptions extends DataSourceJsonData>(
+  datasource: DataSourceApi<TQuery, TOptions>,
+  logsSampleRequest: DataQueryRequest<TQuery>
+): Observable<DataQueryResponse> {
+  logsSampleRequest.hideFromInspector = true;
+
+  return new Observable((observer) => {
+    let rawLogsSample: DataFrame[] = [];
+    observer.next({
+      state: LoadingState.Loading,
+      error: undefined,
+      data: [],
+    });
+
+    const queryResponse = datasource.query(logsSampleRequest);
+    const queryObservable = isObservable(queryResponse) ? queryResponse : from(queryResponse);
+
+    const subscription = queryObservable.subscribe({
+      complete: () => {
+        observer.next({
+          state: LoadingState.Done,
+          error: undefined,
+          data: rawLogsSample,
+        });
+        observer.complete();
+      },
+      next: (dataQueryResponse: DataQueryResponse) => {
+        const { error } = dataQueryResponse;
+        if (error !== undefined) {
+          observer.next({
+            state: LoadingState.Error,
+            error,
+            data: [],
+          });
+          observer.error(error);
+        } else {
+          rawLogsSample = rawLogsSample.concat(dataQueryResponse.data.map(toDataFrame));
+        }
+      },
+      error: (error) => {
+        observer.next({
+          state: LoadingState.Error,
+          error: error,
+          data: [],
+        });
+        observer.error(error);
+      },
+    });
+    return () => {
+      subscription?.unsubscribe();
+    };
+  });
+}
+
 function getIntervalInfo(scopedVars: ScopedVars, timespanMs: number): { interval: string; intervalMs?: number } {
   if (scopedVars.__interval) {
     let intervalMs: number = scopedVars.__interval_ms.value;