LogsVolume: Make log volume work with chunking (#63181)

* aggregate logs volume on streaming

* enable log volume queries to be split

* add streaming indicator

* added tests

* also chunk logs queries

* change extraInfo back to undefined

public/app/core/logsModel.test.ts

@@ -1136,14 +1136,13 @@ describe('logs volume', () => {
     });
   }
 
-  function createExpectedFields(levelName: string, timestamps: number[], values: number[]) {
+  function createExpectedFields(levelName: string) {
     return [
-      { name: 'Time', values: { buffer: timestamps } },
-      {
+      expect.objectContaining({ name: 'Time' }),
+      expect.objectContaining({
         name: 'Value',
-        config: { displayNameFromDS: levelName },
-        values: { buffer: values },
-      },
+        config: expect.objectContaining({ displayNameFromDS: levelName }),
+      }),
     ];
   }
 
@@ -1186,6 +1185,24 @@ describe('logs volume', () => {
     ]);
   }
 
+  function setupMultipleResultsStreaming() {
+    // level=unknown
+    const resultAFrame1 = createFrame({ app: 'app01' }, [100, 200, 300], [5, 5, 5]);
+    // level=error
+    const resultAFrame2 = createFrame({ app: 'app01', level: 'error' }, [100, 200, 300], [0, 1, 0]);
+
+    datasource = new MockObservableDataSourceApi('loki', [
+      {
+        state: LoadingState.Streaming,
+        data: [resultAFrame1],
+      },
+      {
+        state: LoadingState.Streaming,
+        data: [resultAFrame1, resultAFrame2],
+      },
+    ]);
+  }
+
   function setupErrorResponse() {
     datasource = new MockObservableDataSourceApi('loki', [], undefined, 'Error message');
   }
@@ -1194,36 +1211,69 @@ describe('logs volume', () => {
     setup(setupMultipleResults);
 
     await expect(volumeProvider).toEmitValuesWith((received) => {
-      expect(received).toMatchObject([
-        { state: LoadingState.Loading, error: undefined, data: [] },
-        {
-          state: LoadingState.Done,
-          error: undefined,
-          data: [
-            {
-              fields: expect.anything(),
-              meta: {
-                custom: {
-                  targets: [
-                    {
-                      target: 'volume query 1',
-                    },
-                    {
-                      target: 'volume query 2',
-                    },
-                  ],
-                  logsVolumeType: LogsVolumeType.FullRange,
-                  absoluteRange: {
-                    from: FROM.valueOf(),
-                    to: TO.valueOf(),
+      expect(received).toContainEqual({ state: LoadingState.Loading, error: undefined, data: [] });
+      expect(received).toContainEqual({
+        state: LoadingState.Done,
+        error: undefined,
+        data: [
+          expect.objectContaining({
+            fields: expect.anything(),
+            meta: {
+              custom: {
+                targets: [
+                  {
+                    target: 'volume query 1',
+                  },
+                  {
+                    target: 'volume query 2',
                   },
+                ],
+                logsVolumeType: LogsVolumeType.FullRange,
+                absoluteRange: {
+                  from: FROM.valueOf(),
+                  to: TO.valueOf(),
                 },
               },
             },
-            expect.anything(),
-          ],
-        },
-      ]);
+          }),
+          expect.anything(),
+        ],
+      });
+    });
+  });
+
+  it('applies correct meta datya when streaming', async () => {
+    setup(setupMultipleResultsStreaming);
+
+    await expect(volumeProvider).toEmitValuesWith((received) => {
+      expect(received).toContainEqual({ state: LoadingState.Loading, error: undefined, data: [] });
+      expect(received).toContainEqual({
+        state: LoadingState.Done,
+        error: undefined,
+        data: [
+          expect.objectContaining({
+            fields: expect.anything(),
+            meta: {
+              custom: {
+                targets: [
+                  {
+                    target: 'volume query 1',
+                  },
+                  {
+                    target: 'volume query 2',
+                  },
+                ],
+                logsVolumeType: LogsVolumeType.FullRange,
+                absoluteRange: {
+                  from: FROM.valueOf(),
+                  to: TO.valueOf(),
+                },
+              },
+            },
+          }),
+          expect.anything(),
+        ],
+      });
     });
   });
 
@@ -1231,21 +1281,15 @@ describe('logs volume', () => {
     setup(setupMultipleResults);
 
     await expect(volumeProvider).toEmitValuesWith((received) => {
-      expect(received).toMatchObject([
-        { state: LoadingState.Loading, error: undefined, data: [] },
-        {
-          state: LoadingState.Done,
-          error: undefined,
-          data: [
-            {
-              fields: createExpectedFields('unknown', [100, 200, 300], [6, 7, 8]),
-            },
-            {
-              fields: createExpectedFields('error', [100, 200, 300], [1, 2, 1]),
-            },
-          ],
-        },
-      ]);
+      expect(received).toContainEqual({
+        state: LoadingState.Done,
+        error: undefined,
+        data: expect.arrayContaining([
+          expect.objectContaining({
+            fields: expect.arrayContaining(createExpectedFields('error')),
+          }),
+        ]),
+      });
     });
   });
 

public/app/core/logsModel.ts

@@ -706,20 +706,10 @@ export function queryLogsVolume<TQuery extends DataQuery, TOptions extends DataS
 
     const subscription = queryObservable.subscribe({
       complete: () => {
-        const aggregatedLogsVolume = aggregateRawLogsVolume(rawLogsVolume, options.extractLevel);
-        if (aggregatedLogsVolume[0]) {
-          aggregatedLogsVolume[0].meta = {
-            custom: {
-              targets: options.targets,
-              logsVolumeType: LogsVolumeType.FullRange,
-              absoluteRange: { from: options.range.from.valueOf(), to: options.range.to.valueOf() },
-            },
-          };
-        }
         observer.next({
           state: LoadingState.Done,
           error: undefined,
-          data: aggregatedLogsVolume,
+          data: rawLogsVolume,
         });
         observer.complete();
       },
@@ -733,7 +723,25 @@ export function queryLogsVolume<TQuery extends DataQuery, TOptions extends DataS
           });
           observer.error(error);
         } else {
-          rawLogsVolume = rawLogsVolume.concat(dataQueryResponse.data.map(toDataFrame));
+          const aggregatedLogsVolume = aggregateRawLogsVolume(
+            dataQueryResponse.data.map(toDataFrame),
+            options.extractLevel
+          );
+          if (aggregatedLogsVolume[0]) {
+            aggregatedLogsVolume[0].meta = {
+              custom: {
+                targets: options.targets,
+                logsVolumeType: LogsVolumeType.FullRange,
+                absoluteRange: { from: options.range.from.valueOf(), to: options.range.to.valueOf() },
+              },
+            };
+          }
+          rawLogsVolume = aggregatedLogsVolume;
+          observer.next({
+            state: dataQueryResponse.state ?? LoadingState.Streaming,
+            error: undefined,
+            data: rawLogsVolume,
+          });
         }
       },
       error: (error) => {

public/app/features/explore/LogsVolumePanel.test.tsx

@@ -69,4 +69,14 @@ describe('LogsVolumePanel', () => {
     renderPanel(undefined);
     expect(screen.queryByText('Log volume')).not.toBeInTheDocument();
   });
+
+  it('renders a loading indicator when data is streaming', () => {
+    renderPanel({ state: LoadingState.Streaming, error: undefined, data: [{}] });
+    expect(screen.getByTestId('logs-volume-streaming')).toBeInTheDocument();
+  });
+
+  it('does not render loading indicator when data is not streaming', () => {
+    renderPanel({ state: LoadingState.Done, error: undefined, data: [{}] });
+    expect(screen.queryByText('logs-volume-streaming')).not.toBeInTheDocument();
+  });
 });

public/app/features/explore/LogsVolumePanel.tsx

@@ -11,7 +11,7 @@ import {
   EventBus,
   LogsVolumeType,
 } from '@grafana/data';
-import { Button, Collapse, InlineField, TooltipDisplayMode, useStyles2, useTheme2 } from '@grafana/ui';
+import { Button, Collapse, Icon, InlineField, Tooltip, TooltipDisplayMode, useStyles2, useTheme2 } from '@grafana/ui';
 
 import { ExploreGraph } from './Graph/ExploreGraph';
 import { SupplementaryResultError } from './SupplementaryResultError';
@@ -57,7 +57,7 @@ export function LogsVolumePanel(props: Props) {
       LogsVolumePanelContent = (
         <ExploreGraph
           graphStyle="lines"
-          loadingState={LoadingState.Done}
+          loadingState={logsVolumeData.state ?? LoadingState.Done}
           data={logsVolumeData.data}
           height={height}
           width={width - spacing * 2}
@@ -94,6 +94,16 @@ export function LogsVolumePanel(props: Props) {
       </div>
     );
   }
+  if (logsVolumeData.state === LoadingState.Streaming) {
+    extraInfo = (
+      <>
+        {extraInfo}
+        <Tooltip content="Streaming">
+          <Icon name="circle-mono" size="md" className={styles.streaming} data-testid="logs-volume-streaming" />
+        </Tooltip>
+      </>
+    );
+  }
   return (
     <Collapse label="" isOpen={true}>
       <div style={{ height }} className={styles.contentContainer}>
@@ -122,6 +132,9 @@ const getStyles = (theme: GrafanaTheme2) => {
       font-size: ${theme.typography.size.sm};
       color: ${theme.colors.text.secondary};
     `,
+    streaming: css`
+      color: ${theme.colors.success.text};
+    `,
   };
 };
 

public/app/plugins/datasource/loki/queryUtils.ts

@@ -305,13 +305,6 @@ export function requestSupportsPartitioning(queries: LokiQuery[]) {
     return false;
   }
 
-  /**
-   * Disable logs volume queries.
-   */
-  if (queries[0].refId.includes('log-volume-')) {
-    return false;
-  }
-
   return true;
 }