Correlations: Create paginated API (#65241)

* Add pagination params and apply to sql

* Create getCorrelationsResponse that returns metadata

* Set up pagination, change correlations fetch to only get source datasource correlations

* Move correlations from root to pane, only fetch correlations for one datasource when initialized or datasource is changed

* Fix tests

* Fix remaining tests

* Use functional component to handle state

* Remove unneeded mocks, fix tests

* Change perPage to limit

* Fix Go Tests

* Fix linter

* Remove parameter

* Account for mixed datasources

* Delete unused hook

* add source UID filter to API, start backing out front end hook changes

* add source IDs to API, use when loading or changing datasource

* Fix prettier

* Mock correlations response

* Get correlations for all datasources in mixed scenario

* Add documentation for new parameters

* Attempt to fix swagger

* Fix correlations page

* add swagger and openapi docs

* Add mocks to failing test

* Change API for consistency, remove extra hooks and unused function

* Add max to limit and re-gen api docs

* Move the page to the previous page if deleting all the rows on the page

* Only fetch if remove does not have value

* Change page to a reference hook

* Fix documentation, a test and some logic thinking page could be 0

docs/sources/developers/http_api/correlations.md

@@ -267,6 +267,12 @@ Status codes:
 
 Get all correlations.
 
+Query parameters:
+
+- **page** - Optional. Specify which page number to return. Use the limit parameter to specify the number of correlations per page. The default is page 1.
+- **limit** - Optional. Limits the number of returned correlations per page. The default is 100 correlations per page. The maximum limit is 1000 correlations in a page.
+- **sourceUID** - Optional. Specify a source datasource UID to filter by. This can be repeated to filter by multiple datasources.
+
 **Example request:**
 
 ```http

packages/grafana-ui/src/components/Pagination/Pagination.tsx

@@ -102,7 +102,7 @@ export const Pagination = ({
       <ol>
         <li className={styles.item}>
           <Button
-            aria-label="previous"
+            aria-label={`previous page`}
             size="sm"
             variant="secondary"
             onClick={() => onNavigate(currentPage - 1)}
@@ -114,7 +114,7 @@ export const Pagination = ({
         {pageButtons}
         <li className={styles.item}>
           <Button
-            aria-label="next"
+            aria-label={`next page`}
             size="sm"
             variant="secondary"
             onClick={() => onNavigate(currentPage + 1)}

pkg/services/correlations/api.go

@@ -76,7 +76,7 @@ type CreateCorrelationParams struct {
 	SourceUID string `json:"sourceUID"`
 }
 
-//swagger:response createCorrelationResponse
+// swagger:response createCorrelationResponse
 type CreateCorrelationResponse struct {
 	// in: body
 	Body CreateCorrelationResponseBody `json:"body"`
@@ -192,7 +192,7 @@ type UpdateCorrelationParams struct {
 	Body UpdateCorrelationCommand `json:"body"`
 }
 
-//swagger:response updateCorrelationResponse
+// swagger:response updateCorrelationResponse
 type UpdateCorrelationResponse struct {
 	// in: body
 	Body UpdateCorrelationResponseBody `json:"body"`
@@ -282,7 +282,7 @@ type GetCorrelationsBySourceUIDParams struct {
 	DatasourceUID string `json:"sourceUID"`
 }
 
-//swagger:response getCorrelationsBySourceUIDResponse
+// swagger:response getCorrelationsBySourceUIDResponse
 type GetCorrelationsBySourceUIDResponse struct {
 	// in: body
 	Body []Correlation `json:"body"`
@@ -298,8 +298,25 @@ type GetCorrelationsBySourceUIDResponse struct {
 // 404: notFoundError
 // 500: internalServerError
 func (s *CorrelationsService) getCorrelationsHandler(c *contextmodel.ReqContext) response.Response {
+	limit := c.QueryInt64("limit")
+	if limit <= 0 {
+		limit = 100
+	} else if limit > 1000 {
+		limit = 1000
+	}
+
+	page := c.QueryInt64("page")
+	if page <= 0 {
+		page = 1
+	}
+
+	sourceUIDs := c.QueryStrings("sourceUID")
+
 	query := GetCorrelationsQuery{
-		OrgId: c.OrgID,
+		OrgId:      c.OrgID,
+		Limit:      limit,
+		Page:       page,
+		SourceUIDs: sourceUIDs,
 	}
 
 	correlations, err := s.getCorrelations(c.Req.Context(), query)
@@ -314,6 +331,27 @@ func (s *CorrelationsService) getCorrelationsHandler(c *contextmodel.ReqContext)
 	return response.JSON(http.StatusOK, correlations)
 }
 
+// swagger:parameters getCorrelations
+type GetCorrelationsParams struct {
+	// Limit the maximum number of correlations to return per page
+	// in:query
+	// required:false
+	// default:100
+	// maximum: 1000
+	Limit int64 `json:"limit"`
+	// Page index for starting fetching correlations
+	// in:query
+	// required:false
+	// default:1
+	Page int64 `json:"page"`
+	// Source datasource UID filter to be applied to correlations
+	// in:query
+	// type: array
+	// collectionFormat: multi
+	// required:false
+	SourceUIDs []string `json:"sourceUID"`
+}
+
 //swagger:response getCorrelationsResponse
 type GetCorrelationsResponse struct {
 	// in: body

pkg/services/correlations/correlations.go

@@ -94,7 +94,7 @@ func (s CorrelationsService) GetCorrelationsBySourceUID(ctx context.Context, cmd
 	return s.getCorrelationsBySourceUID(ctx, cmd)
 }
 
-func (s CorrelationsService) GetCorrelations(ctx context.Context, cmd GetCorrelationsQuery) ([]Correlation, error) {
+func (s CorrelationsService) GetCorrelations(ctx context.Context, cmd GetCorrelationsQuery) (GetCorrelationsResponseBody, error) {
 	return s.getCorrelations(ctx, cmd)
 }
 

pkg/services/correlations/database.go

@@ -225,17 +225,42 @@ func (s CorrelationsService) getCorrelationsBySourceUID(ctx context.Context, cmd
 	return correlations, nil
 }
 
-func (s CorrelationsService) getCorrelations(ctx context.Context, cmd GetCorrelationsQuery) ([]Correlation, error) {
-	correlations := make([]Correlation, 0)
-
-	err := s.SQLStore.WithDbSession(ctx, func(session *db.Session) error {
-		return session.Select("correlation.*").Join("", "data_source AS dss", "correlation.source_uid = dss.uid and dss.org_id = ?", cmd.OrgId).Join("", "data_source AS dst", "correlation.target_uid = dst.uid and dst.org_id = ?", cmd.OrgId).Find(&correlations)
-	})
-	if err != nil {
-		return []Correlation{}, err
+func (s CorrelationsService) getCorrelations(ctx context.Context, cmd GetCorrelationsQuery) (GetCorrelationsResponseBody, error) {
+	result := GetCorrelationsResponseBody{
+		Correlations: make([]Correlation, 0),
+		Page:         cmd.Page,
+		Limit:        cmd.Limit,
 	}
 
-	return correlations, nil
+	err := s.SQLStore.WithDbSession(ctx, func(session *db.Session) error {
+		offset := cmd.Limit * (cmd.Page - 1)
+
+		q := session.Select("correlation.*").Join("", "data_source AS dss", "correlation.source_uid = dss.uid and dss.org_id = ?", cmd.OrgId).Join("", "data_source AS dst", "correlation.target_uid = dst.uid and dst.org_id = ?", cmd.OrgId)
+
+		if len(cmd.SourceUIDs) > 0 {
+			q.In("dss.uid", cmd.SourceUIDs)
+		}
+
+		return q.Limit(int(cmd.Limit), int(offset)).Find(&result.Correlations)
+	})
+	if err != nil {
+		return GetCorrelationsResponseBody{}, err
+	}
+
+	count, err := s.CountCorrelations(ctx)
+	if err != nil {
+		return GetCorrelationsResponseBody{}, err
+	}
+
+	tag, err := quota.NewTag(QuotaTargetSrv, QuotaTarget, quota.GlobalScope)
+	if err != nil {
+		return GetCorrelationsResponseBody{}, err
+	}
+
+	totalCount, _ := count.Get(tag)
+	result.TotalCount = totalCount
+
+	return result, nil
 }
 
 func (s CorrelationsService) deleteCorrelationsBySourceUID(ctx context.Context, cmd DeleteCorrelationsBySourceUIDCommand) error {

pkg/services/correlations/models.go

@@ -73,11 +73,11 @@ type CorrelationConfig struct {
 	Type CorrelationConfigType `json:"type" binding:"Required"`
 	// Target data query
 	// required:true
-	// example: { "expr": "job=app" }
+	// example: {"prop1":"value1","prop2":"value"}
 	Target map[string]interface{} `json:"target" binding:"Required"`
 	// Source data transformations
 	// required:false
-	// example: [{"type": "logfmt"}]
+	// example: [{"type":"logfmt"}]
 	Transformations Transformations `json:"transformations,omitempty"`
 }
 
@@ -107,10 +107,10 @@ type Correlation struct {
 	// example: 50xhMlg9k
 	UID string `json:"uid" xorm:"pk 'uid'"`
 	// UID of the data source the correlation originates from
-	// example:d0oxYRg4z
+	// example: d0oxYRg4z
 	SourceUID string `json:"sourceUID" xorm:"pk 'source_uid'"`
 	// UID of the data source the correlation points to
-	// example:PE1C5CBDA0504A6A3
+	// example: PE1C5CBDA0504A6A3
 	TargetUID *string `json:"targetUID" xorm:"target_uid"`
 	// Label identifying the correlation
 	// example: My Label
@@ -122,6 +122,13 @@ type Correlation struct {
 	Config CorrelationConfig `json:"config" xorm:"jsonb config"`
 }
 
+type GetCorrelationsResponseBody struct {
+	Correlations []Correlation `json:"correlations"`
+	TotalCount   int64         `json:"totalCount"`
+	Page         int64         `json:"page"`
+	Limit        int64         `json:"limit"`
+}
+
 // CreateCorrelationResponse is the response struct for CreateCorrelationCommand
 // swagger:model
 type CreateCorrelationResponseBody struct {
@@ -138,7 +145,7 @@ type CreateCorrelationCommand struct {
 	OrgId             int64  `json:"-"`
 	SkipReadOnlyCheck bool   `json:"-"`
 	// Target data source UID to which the correlation is created. required if config.type = query
-	// example:PE1C5CBDA0504A6A3
+	// example: PE1C5CBDA0504A6A3
 	TargetUID *string `json:"targetUID"`
 	// Optional label identifying the correlation
 	// example: My label
@@ -193,7 +200,7 @@ type CorrelationConfigUpdateDTO struct {
 	// Target type
 	Type *CorrelationConfigType `json:"type"`
 	// Target data query
-	// example: { "expr": "job=app" }
+	// example: {"prop1":"value1","prop2":"value"}
 	Target *map[string]interface{} `json:"target"`
 	// Source data transformations
 	// example: [{"type": "logfmt"},{"type":"regex","expression":"(Superman|Batman)", "variable":"name"}]
@@ -260,6 +267,21 @@ type GetCorrelationsBySourceUIDQuery struct {
 // GetCorrelationsQuery is the query to retrieve all correlations
 type GetCorrelationsQuery struct {
 	OrgId int64 `json:"-"`
+	// Limit the maximum number of correlations to return per page
+	// in:query
+	// required:false
+	// default:100
+	Limit int64 `json:"limit"`
+	// Page index for starting fetching correlations
+	// in:query
+	// required:false
+	// default:1
+	Page int64 `json:"page"`
+
+	// Source datasource UID filter to be applied to correlations
+	// in:query
+	// required:false
+	SourceUIDs []string `json:"sourceuid"`
 }
 
 type DeleteCorrelationsBySourceUIDCommand struct {

pkg/tests/api/correlations/common_test.go

@@ -51,12 +51,13 @@ type User struct {
 type GetParams struct {
 	url  string
 	user User
+	page string
 }
 
 func (c TestContext) Get(params GetParams) *http.Response {
 	c.t.Helper()
-
-	resp, err := http.Get(c.getURL(params.url, params.user))
+	fmtUrl := fmt.Sprintf("%s?page=%s", params.url, params.page)
+	resp, err := http.Get(c.getURL(fmtUrl, params.user))
 	require.NoError(c.t, err)
 
 	return resp

pkg/tests/api/correlations/correlations_read_test.go

@@ -42,17 +42,18 @@ func TestIntegrationReadCorrelation(t *testing.T) {
 			res := ctx.Get(GetParams{
 				url:  "/api/datasources/correlations",
 				user: adminUser,
+				page: "0",
 			})
 			require.Equal(t, http.StatusOK, res.StatusCode)
 
 			responseBody, err := io.ReadAll(res.Body)
 			require.NoError(t, err)
 
-			var response []correlations.Correlation
+			var response correlations.GetCorrelationsResponseBody
 			err = json.Unmarshal(responseBody, &response)
 			require.NoError(t, err)
 
-			require.Len(t, response, 0)
+			require.Len(t, response.Correlations, 0)
 
 			require.NoError(t, res.Body.Close())
 		})
@@ -147,12 +148,12 @@ func TestIntegrationReadCorrelation(t *testing.T) {
 			responseBody, err := io.ReadAll(res.Body)
 			require.NoError(t, err)
 
-			var response []correlations.Correlation
+			var response correlations.GetCorrelationsResponseBody
 			err = json.Unmarshal(responseBody, &response)
 			require.NoError(t, err)
 
-			require.Len(t, response, 1)
-			require.EqualValues(t, correlation, response[0])
+			require.Len(t, response.Correlations, 1)
+			require.EqualValues(t, correlation, response.Correlations[0])
 
 			require.NoError(t, res.Body.Close())
 		})

public/app/features/correlations/CorrelationsPage.test.tsx

@@ -73,7 +73,14 @@ const renderWithContext = async (
     },
     fetch: (options: BackendSrvRequest) => {
       return new Observable((s) => {
-        s.next(merge(createFetchCorrelationsResponse({ url: options.url, data: correlations })));
+        s.next(
+          merge(
+            createFetchCorrelationsResponse({
+              url: options.url,
+              data: { correlations, page: 1, limit: 5, totalCount: 0 },
+            })
+          )
+        );
         s.complete();
       });
     },

public/app/features/correlations/CorrelationsPage.tsx

@@ -1,6 +1,6 @@
 import { css } from '@emotion/css';
 import { negate } from 'lodash';
-import React, { memo, useCallback, useEffect, useMemo, useState } from 'react';
+import React, { memo, useCallback, useEffect, useMemo, useRef, useState } from 'react';
 
 import { GrafanaTheme2 } from '@grafana/data';
 import { isFetchError, reportInteraction } from '@grafana/runtime';
@@ -15,6 +15,7 @@ import {
   type Column,
   type CellProps,
   type SortByFn,
+  Pagination,
   Icon,
 } from '@grafana/ui';
 import { Page } from 'app/core/components/Page/Page';
@@ -41,6 +42,7 @@ const loaderWrapper = css`
 export default function CorrelationsPage() {
   const navModel = useNavModel('correlations');
   const [isAdding, setIsAddingValue] = useState(false);
+  const page = useRef(1);
 
   const setIsAdding = (value: boolean) => {
     setIsAddingValue(value);
@@ -54,61 +56,59 @@ export default function CorrelationsPage() {
     get: { execute: fetchCorrelations, ...get },
   } = useCorrelations();
 
-  useEffect(() => {
-    fetchCorrelations();
-    // we only want to fetch data on first render
-    // eslint-disable-next-line react-hooks/exhaustive-deps
-  }, []);
-
   const canWriteCorrelations = contextSrv.hasPermission(AccessControlAction.DataSourcesWrite);
 
   const handleAdded = useCallback(() => {
     reportInteraction('grafana_correlations_added');
-    fetchCorrelations();
+    fetchCorrelations({ page: page.current });
     setIsAdding(false);
   }, [fetchCorrelations]);
 
   const handleUpdated = useCallback(() => {
     reportInteraction('grafana_correlations_edited');
-    fetchCorrelations();
+    fetchCorrelations({ page: page.current });
   }, [fetchCorrelations]);
 
   const handleDelete = useCallback(
-    (params: RemoveCorrelationParams) => {
-      remove.execute(params);
+    async (params: RemoveCorrelationParams, isLastRow: boolean) => {
+      await remove.execute(params);
+      reportInteraction('grafana_correlations_deleted');
+
+      if (isLastRow) {
+        page.current--;
+      }
+      fetchCorrelations({ page: page.current });
     },
-    [remove]
+    [remove, fetchCorrelations]
   );
 
-  // onDelete - triggers when deleting a correlation
   useEffect(() => {
-    if (remove.value) {
-      reportInteraction('grafana_correlations_deleted');
-    }
-  }, [remove.value]);
-
-  useEffect(() => {
-    if (!remove.error && !remove.loading && remove.value) {
-      fetchCorrelations();
-    }
-  }, [remove.error, remove.loading, remove.value, fetchCorrelations]);
+    fetchCorrelations({ page: page.current });
+  }, [fetchCorrelations]);
 
   const RowActions = useCallback(
     ({
       row: {
+        index,
         original: {
           source: { uid: sourceUID, readOnly },
           uid,
         },
       },
-    }: CellProps<CorrelationData, void>) =>
-      !readOnly && (
-        <DeleteButton
-          aria-label="delete correlation"
-          onConfirm={() => handleDelete({ sourceUID, uid })}
-          closeOnConfirm
-        />
-      ),
+    }: CellProps<CorrelationData, void>) => {
+      return (
+        !readOnly && (
+          <DeleteButton
+            aria-label="delete correlation"
+            onConfirm={() =>
+              handleDelete({ sourceUID, uid }, page.current > 1 && index === 0 && data?.correlations.length === 1)
+            }
+            closeOnConfirm
+          />
+        )
+      );
+    },
+    // eslint-disable-next-line react-hooks/exhaustive-deps
     [handleDelete]
   );
 
@@ -144,8 +144,8 @@ export default function CorrelationsPage() {
   );
 
   const data = useMemo(() => get.value, [get.value]);
-  const showEmptyListCTA = data?.length === 0 && !isAdding && !get.error;
-  const addButton = canWriteCorrelations && data?.length !== 0 && data !== undefined && !isAdding && (
+  const showEmptyListCTA = data?.correlations.length === 0 && !isAdding && !get.error;
+  const addButton = canWriteCorrelations && data?.correlations?.length !== 0 && data !== undefined && !isAdding && (
     <Button icon="plus" onClick={() => setIsAdding(true)}>
       Add new
     </Button>
@@ -188,19 +188,28 @@ export default function CorrelationsPage() {
 
           {isAdding && <AddCorrelationForm onClose={() => setIsAdding(false)} onCreated={handleAdded} />}
 
-          {data && data.length >= 1 && (
-            <InteractiveTable
-              renderExpandedRow={(correlation) => (
-                <ExpendedRow
-                  correlation={correlation}
-                  onUpdated={handleUpdated}
-                  readOnly={isSourceReadOnly({ source: correlation.source }) || !canWriteCorrelations}
-                />
-              )}
-              columns={columns}
-              data={data}
-              getRowId={(correlation) => `${correlation.source.uid}-${correlation.uid}`}
-            />
+          {data && data.correlations.length >= 1 && (
+            <>
+              <InteractiveTable
+                renderExpandedRow={(correlation) => (
+                  <ExpendedRow
+                    correlation={correlation}
+                    onUpdated={handleUpdated}
+                    readOnly={isSourceReadOnly({ source: correlation.source }) || !canWriteCorrelations}
+                  />
+                )}
+                columns={columns}
+                data={data.correlations}
+                getRowId={(correlation) => `${correlation.source.uid}-${correlation.uid}`}
+              />
+              <Pagination
+                currentPage={page.current}
+                numberOfPages={Math.ceil(data.totalCount / data.limit)}
+                onNavigate={(toPage: number) => {
+                  fetchCorrelations({ page: (page.current = toPage) });
+                }}
+              />
+            </>
           )}
         </div>
       </Page.Contents>

public/app/features/correlations/types.ts

@@ -44,6 +44,10 @@ export interface Correlation {
   config: CorrelationConfig;
 }
 
+export type GetCorrelationsParams = {
+  page: number;
+};
+
 export type RemoveCorrelationParams = Pick<Correlation, 'sourceUID' | 'uid'>;
 export type CreateCorrelationParams = Omit<Correlation, 'uid'>;
 export type UpdateCorrelationParams = Omit<Correlation, 'targetUID'>;

public/app/features/correlations/useCorrelations.ts

@@ -9,17 +9,32 @@ import {
   Correlation,
   CreateCorrelationParams,
   CreateCorrelationResponse,
+  GetCorrelationsParams,
   RemoveCorrelationParams,
   RemoveCorrelationResponse,
   UpdateCorrelationParams,
   UpdateCorrelationResponse,
 } from './types';
 
+export interface CorrelationsResponse {
+  correlations: Correlation[];
+  page: number;
+  limit: number;
+  totalCount: number;
+}
+
 export interface CorrelationData extends Omit<Correlation, 'sourceUID' | 'targetUID'> {
   source: DataSourceInstanceSettings;
   target: DataSourceInstanceSettings;
 }
 
+export interface CorrelationsData {
+  correlations: CorrelationData[];
+  page: number;
+  limit: number;
+  totalCount: number;
+}
+
 const toEnrichedCorrelationData = ({
   sourceUID,
   targetUID,
@@ -39,10 +54,14 @@ const toEnrichedCorrelationData = ({
 
 const validSourceFilter = (correlation: CorrelationData | undefined): correlation is CorrelationData => !!correlation;
 
-const toEnrichedCorrelationsData = (correlations: Correlation[]): CorrelationData[] => {
-  return correlations.map(toEnrichedCorrelationData).filter(validSourceFilter);
+export const toEnrichedCorrelationsData = (correlationsResponse: CorrelationsResponse): CorrelationsData => {
+  return {
+    ...correlationsResponse,
+    correlations: correlationsResponse.correlations.map(toEnrichedCorrelationData).filter(validSourceFilter),
+  };
 };
-function getData<T>(response: FetchResponse<T>) {
+
+export function getData<T>(response: FetchResponse<T>) {
   return response.data;
 }
 
@@ -55,10 +74,15 @@ function getData<T>(response: FetchResponse<T>) {
 export const useCorrelations = () => {
   const { backend } = useGrafana();
 
-  const [getInfo, get] = useAsyncFn<() => Promise<CorrelationData[]>>(
-    () =>
+  const [getInfo, get] = useAsyncFn<(params: GetCorrelationsParams) => Promise<CorrelationsData>>(
+    (params) =>
       lastValueFrom(
-        backend.fetch<Correlation[]>({ url: '/api/datasources/correlations', method: 'GET', showErrorAlert: false })
+        backend.fetch<CorrelationsResponse>({
+          url: '/api/datasources/correlations',
+          params: { page: params.page },
+          method: 'GET',
+          showErrorAlert: false,
+        })
       )
         .then(getData)
         .then(toEnrichedCorrelationsData),

public/app/features/correlations/utils.ts

@@ -1,8 +1,17 @@
+import { lastValueFrom } from 'rxjs';
+
 import { DataFrame, DataLinkConfigOrigin } from '@grafana/data';
+import { getBackendSrv } from '@grafana/runtime';
 
 import { formatValueName } from '../explore/PrometheusListView/ItemLabels';
 
-import { CorrelationData } from './useCorrelations';
+import {
+  CorrelationData,
+  CorrelationsData,
+  CorrelationsResponse,
+  getData,
+  toEnrichedCorrelationsData,
+} from './useCorrelations';
 
 type DataFrameRefIdToDataSourceUid = Record<string, string>;
 
@@ -58,3 +67,18 @@ const decorateDataFrameWithInternalDataLinks = (dataFrame: DataFrame, correlatio
     });
   });
 };
+
+export const getCorrelationsBySourceUIDs = async (sourceUIDs: string[]): Promise<CorrelationsData> => {
+  return lastValueFrom(
+    getBackendSrv().fetch<CorrelationsResponse>({
+      url: `/api/datasources/correlations`,
+      method: 'GET',
+      showErrorAlert: false,
+      params: {
+        sourceUID: sourceUIDs,
+      },
+    })
+  )
+    .then(getData)
+    .then(toEnrichedCorrelationsData);
+};

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

@@ -31,6 +31,20 @@ jest.mock('react-virtualized-auto-sizer', () => {
   };
 });
 
+const fetch = jest.fn().mockResolvedValue({ correlations: [] });
+jest.mock('@grafana/runtime', () => ({
+  ...jest.requireActual('@grafana/runtime'),
+  getBackendSrv: () => ({ fetch }),
+}));
+
+jest.mock('rxjs', () => ({
+  ...jest.requireActual('rxjs'),
+  lastValueFrom: () =>
+    new Promise((resolve, reject) => {
+      resolve({ data: { correlations: [] } });
+    }),
+}));
+
 describe('ExplorePage', () => {
   afterEach(() => {
     tearDown();

public/app/features/explore/ExplorePage.tsx

@@ -13,7 +13,6 @@ import { ExploreQueryParams } from 'app/types/explore';
 
 import { ExploreActions } from './ExploreActions';
 import { ExplorePaneContainer } from './ExplorePaneContainer';
-import { useExploreCorrelations } from './hooks/useExploreCorrelations';
 import { useExplorePageTitle } from './hooks/useExplorePageTitle';
 import { useStateSync } from './hooks/useStateSync';
 import { useTimeSrvFix } from './hooks/useTimeSrvFix';
@@ -39,7 +38,6 @@ export default function ExplorePage(props: GrafanaRouteComponentProps<{}, Explor
   // if we were to update the URL on state change, the title would not match the URL.
   // Ultimately the URL is the single source of truth from which state is derived, the page title is not different
   useExplorePageTitle(props.queryParams);
-  useExploreCorrelations();
   const dispatch = useDispatch();
   const { keybindings, chrome } = useGrafana();
   const navModel = useNavModel('explore');

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

@@ -56,10 +56,10 @@ function setup(queries: DataQuery[]) {
         richHistory: [],
         datasourceInstance: datasources['someDs-uid'],
         queries,
+        correlations: [],
       },
     },
     syncedTimes: false,
-    correlations: [],
     richHistoryStorageFull: false,
     richHistoryLimitExceededWarningShown: false,
   };

public/app/features/explore/hooks/useExploreCorrelations.ts

@@ -1,35 +0,0 @@
-import { useEffect } from 'react';
-
-import { config } from '@grafana/runtime';
-import { useAppNotification } from 'app/core/copy/appNotification';
-import { useCorrelations } from 'app/features/correlations/useCorrelations';
-import { useDispatch } from 'app/types';
-
-import { saveCorrelationsAction } from '../state/main';
-
-export function useExploreCorrelations() {
-  const { get } = useCorrelations();
-  const { warning } = useAppNotification();
-
-  const dispatch = useDispatch();
-  useEffect(() => {
-    if (!config.featureToggles.correlations) {
-      dispatch(saveCorrelationsAction([]));
-    } else {
-      get.execute();
-    }
-    // eslint-disable-next-line react-hooks/exhaustive-deps
-  }, []);
-
-  useEffect(() => {
-    if (get.value) {
-      dispatch(saveCorrelationsAction(get.value));
-    } else if (get.error) {
-      dispatch(saveCorrelationsAction([]));
-      warning(
-        'Could not load correlations.',
-        'Correlations data could not be loaded, DataLinks may have partial data.'
-      );
-    }
-  }, [get.value, get.error, dispatch, warning]);
-}

public/app/features/explore/hooks/useStateSync/index.test.tsx

@@ -17,6 +17,20 @@ import { splitClose, splitOpen } from '../../state/main';
 
 import { useStateSync } from './';
 
+const fetch = jest.fn().mockResolvedValue({ correlations: [] });
+jest.mock('@grafana/runtime', () => ({
+  ...jest.requireActual('@grafana/runtime'),
+  getBackendSrv: () => ({ fetch }),
+}));
+
+jest.mock('rxjs', () => ({
+  ...jest.requireActual('rxjs'),
+  lastValueFrom: () =>
+    new Promise((resolve, reject) => {
+      resolve({ data: { correlations: [] } });
+    }),
+}));
+
 function defaultDsGetter(datasources: Array<ReturnType<typeof makeDatasourceSetup>>): DataSourceSrv['get'] {
   return (datasource) => {
     let ds;

public/app/features/explore/spec/datasourceState.test.tsx

@@ -4,6 +4,12 @@ import { changeDatasource } from './helper/interactions';
 import { makeLogsQueryResponse } from './helper/query';
 import { setupExplore, tearDown, waitForExplore } from './helper/setup';
 
+jest.mock('../../correlations/utils', () => {
+  return {
+    getCorrelationsBySourceUIDs: jest.fn().mockReturnValue({ correlations: [] }),
+  };
+});
+
 describe('Explore: handle datasource states', () => {
   afterEach(() => {
     tearDown();

public/app/features/explore/spec/interpolation.test.tsx

@@ -29,6 +29,12 @@ jest.mock('react-virtualized-auto-sizer', () => {
   };
 });
 
+jest.mock('../../correlations/utils', () => {
+  return {
+    getCorrelationsBySourceUIDs: jest.fn().mockReturnValue({ correlations: [] }),
+  };
+});
+
 describe('Explore: interpolation', () => {
   // support-escalations/issues/1459
   it('Time is interpolated when explore is opened with a URL', async () => {

public/app/features/explore/spec/query.test.tsx

@@ -5,6 +5,12 @@ import { serializeStateToUrlParam } from '@grafana/data';
 import { makeLogsQueryResponse } from './helper/query';
 import { setupExplore, tearDown, waitForExplore } from './helper/setup';
 
+jest.mock('../../correlations/utils', () => {
+  return {
+    getCorrelationsBySourceUIDs: jest.fn().mockReturnValue({ correlations: [] }),
+  };
+});
+
 describe('Explore: handle running/not running query', () => {
   afterEach(() => {
     tearDown();

public/app/features/explore/spec/queryHistory.test.tsx

@@ -74,6 +74,12 @@ jest.mock('react-virtualized-auto-sizer', () => {
   };
 });
 
+jest.mock('../../correlations/utils', () => {
+  return {
+    getCorrelationsBySourceUIDs: jest.fn().mockReturnValue({ correlations: [] }),
+  };
+});
+
 describe('Explore: Query History', () => {
   const USER_INPUT = 'my query';
   const RAW_QUERY = `{"expr":"${USER_INPUT}"}`;

public/app/features/explore/state/datasource.ts

@@ -6,13 +6,15 @@ import { reportInteraction } from '@grafana/runtime';
 import { DataSourceRef } from '@grafana/schema';
 import { RefreshPicker } from '@grafana/ui';
 import { stopQueryState } from 'app/core/utils/explore';
+import { getCorrelationsBySourceUIDs } from 'app/features/correlations/utils';
 import { ExploreItemState, ThunkResult } from 'app/types';
 
 import { loadSupplementaryQueries } from '../utils/supplementaryQueries';
 
+import { saveCorrelationsAction } from './explorePane';
 import { importQueries, runQueries } from './query';
 import { changeRefreshInterval } from './time';
-import { createEmptyQueryResponse, loadAndInitDatasource } from './utils';
+import { createEmptyQueryResponse, getDatasourceUIDs, loadAndInitDatasource } from './utils';
 
 //
 // Actions and Payloads
@@ -60,8 +62,13 @@ export function changeDatasource(
       })
     );
 
+    const queries = getState().explore.panes[exploreId]!.queries;
+
+    const datasourceUIDs = getDatasourceUIDs(instance.uid, queries);
+    const correlations = await getCorrelationsBySourceUIDs(datasourceUIDs);
+    dispatch(saveCorrelationsAction({ exploreId: exploreId, correlations: correlations.correlations || [] }));
+
     if (options?.importQueries) {
-      const queries = getState().explore.panes[exploreId]!.queries;
       await dispatch(importQueries(exploreId, queries, currentDataSourceInstance, instance));
     }
 

public/app/features/explore/state/explorePane.ts

@@ -11,6 +11,8 @@ import {
 } from '@grafana/data';
 import { DataQuery, DataSourceRef } from '@grafana/schema';
 import { getQueryKeys } from 'app/core/utils/explore';
+import { CorrelationData } from 'app/features/correlations/useCorrelations';
+import { getCorrelationsBySourceUIDs } from 'app/features/correlations/utils';
 import { getTimeZone } from 'app/features/profile/state/selectors';
 import { createAsyncThunk, ThunkResult } from 'app/types';
 import { ExploreItemState } from 'app/types/explore';
@@ -20,7 +22,13 @@ import { historyReducer } from './history';
 import { richHistorySearchFiltersUpdatedAction, richHistoryUpdatedAction } from './main';
 import { queryReducer, runQueries } from './query';
 import { timeReducer, updateTime } from './time';
-import { makeExplorePaneState, loadAndInitDatasource, createEmptyQueryResponse, getRange } from './utils';
+import {
+  makeExplorePaneState,
+  loadAndInitDatasource,
+  createEmptyQueryResponse,
+  getRange,
+  getDatasourceUIDs,
+} from './utils';
 // Types
 
 //
@@ -86,6 +94,12 @@ export interface SetUrlReplacedPayload {
 }
 export const setUrlReplacedAction = createAction<SetUrlReplacedPayload>('explore/setUrlReplaced');
 
+export interface SaveCorrelationsPayload {
+  exploreId: string;
+  correlations: CorrelationData[];
+}
+export const saveCorrelationsAction = createAction<SaveCorrelationsPayload>('explore/saveCorrelationsAction');
+
 /**
  * Keep track of the Explore container size, in particular the width.
  * The width will be used to calculate graph intervals (number of datapoints).
@@ -141,6 +155,10 @@ export const initializeExplore = createAsyncThunk(
     dispatch(updateTime({ exploreId }));
 
     if (instance) {
+      const datasourceUIDs = getDatasourceUIDs(instance.uid, queries);
+      const correlations = await getCorrelationsBySourceUIDs(datasourceUIDs);
+      dispatch(saveCorrelationsAction({ exploreId: exploreId, correlations: correlations.correlations || [] }));
+
       dispatch(runQueries({ exploreId }));
     }
 
@@ -189,6 +207,13 @@ export const paneReducer = (state: ExploreItemState = makeExplorePaneState(), ac
     return { ...state, panelsState };
   }
 
+  if (saveCorrelationsAction.match(action)) {
+    return {
+      ...state,
+      correlations: action.payload.correlations,
+    };
+  }
+
   if (initializeExploreAction.match(action)) {
     const { queries, range, datasourceInstance, history } = action.payload;
 
@@ -202,6 +227,7 @@ export const paneReducer = (state: ExploreItemState = makeExplorePaneState(), ac
       history,
       queryResponse: createEmptyQueryResponse(),
       cache: [],
+      correlations: [],
     };
   }
 

public/app/features/explore/state/main.ts

@@ -10,7 +10,6 @@ import { ExploreItemState, ExploreState } from 'app/types/explore';
 import { RichHistoryResults } from '../../../core/history/RichHistoryStorage';
 import { RichHistorySearchFilters, RichHistorySettings } from '../../../core/utils/richHistoryTypes';
 import { createAsyncThunk, ThunkResult } from '../../../types';
-import { CorrelationData } from '../../correlations/useCorrelations';
 import { TimeSrv } from '../../dashboard/services/TimeSrv';
 import { withUniqueRefIds } from '../utils/queries';
 
@@ -38,8 +37,6 @@ export const richHistorySearchFiltersUpdatedAction = createAction<{
   filters?: RichHistorySearchFilters;
 }>('explore/richHistorySearchFiltersUpdatedAction');
 
-export const saveCorrelationsAction = createAction<CorrelationData[]>('explore/saveCorrelationsAction');
-
 export const splitSizeUpdateAction = createAction<{
   largerExploreId?: string;
 }>('explore/splitSizeUpdateAction');
@@ -144,7 +141,6 @@ const initialExploreItemState = makeExplorePaneState();
 export const initialExploreState: ExploreState = {
   syncedTimes: false,
   panes: {},
-  correlations: undefined,
   richHistoryStorageFull: false,
   richHistoryLimitExceededWarningShown: false,
   largerExploreId: undefined,
@@ -199,13 +195,6 @@ export const exploreReducer = (state = initialExploreState, action: AnyAction):
     };
   }
 
-  if (saveCorrelationsAction.match(action)) {
-    return {
-      ...state,
-      correlations: action.payload,
-    };
-  }
-
   if (syncTimesAction.match(action)) {
     return { ...state, syncedTimes: action.payload.syncedTimes };
   }

public/app/features/explore/state/query.test.ts

@@ -25,8 +25,8 @@ import { setTimeSrv, TimeSrv } from '../../dashboard/services/TimeSrv';
 import { makeLogs } from '../__mocks__/makeLogs';
 import { supplementaryQueryTypes } from '../utils/supplementaryQueries';
 
+import { saveCorrelationsAction } from './explorePane';
 import { createDefaultInitialState } from './helpers';
-import { saveCorrelationsAction } from './main';
 import {
   addQueryRowAction,
   addResultsToCache,
@@ -159,7 +159,8 @@ describe('runQueries', () => {
   it('should pass dataFrames to state even if there is error in response', async () => {
     const { dispatch, getState } = setupTests();
     setupQueryResponse(getState());
-    await dispatch(saveCorrelationsAction([]));
+
+    await dispatch(saveCorrelationsAction({ exploreId: 'left', correlations: [] }));
     await dispatch(runQueries({ exploreId: 'left' }));
     expect(getState().explore.panes.left!.showMetrics).toBeTruthy();
     expect(getState().explore.panes.left!.graphResult).toBeDefined();
@@ -168,7 +169,7 @@ describe('runQueries', () => {
   it('should modify the request-id for all supplementary queries', () => {
     const { dispatch, getState } = setupTests();
     setupQueryResponse(getState());
-    dispatch(saveCorrelationsAction([]));
+    dispatch(saveCorrelationsAction({ exploreId: 'left', correlations: [] }));
     dispatch(runQueries({ exploreId: 'left' }));
 
     const state = getState().explore.panes.left!;
@@ -188,7 +189,7 @@ describe('runQueries', () => {
     const { dispatch, getState } = setupTests();
     const leftDatasourceInstance = assertIsDefined(getState().explore.panes.left!.datasourceInstance);
     jest.mocked(leftDatasourceInstance.query).mockReturnValueOnce(EMPTY);
-    await dispatch(saveCorrelationsAction([]));
+    await dispatch(saveCorrelationsAction({ exploreId: 'left', correlations: [] }));
     await dispatch(runQueries({ exploreId: 'left' }));
     await new Promise((resolve) => setTimeout(() => resolve(''), 500));
     expect(getState().explore.panes.left!.queryResponse.state).toBe(LoadingState.Done);
@@ -199,7 +200,7 @@ describe('runQueries', () => {
     setupQueryResponse(getState());
     await dispatch(runQueries({ exploreId: 'left' }));
     expect(getState().explore.panes.left!.graphResult).not.toBeDefined();
-    await dispatch(saveCorrelationsAction([]));
+    await dispatch(saveCorrelationsAction({ exploreId: 'left', correlations: [] }));
     expect(getState().explore.panes.left!.graphResult).toBeDefined();
   });
 });

public/app/features/explore/state/query.ts

@@ -35,6 +35,7 @@ import {
 } from 'app/core/utils/explore';
 import { getShiftedTimeRange } from 'app/core/utils/timePicker';
 import { CorrelationData } from 'app/features/correlations/useCorrelations';
+import { getCorrelationsBySourceUIDs } from 'app/features/correlations/utils';
 import { getTimeZone } from 'app/features/profile/state/selectors';
 import { MIXED_DATASOURCE_NAME } from 'app/plugins/datasource/mixed/MixedDataSource';
 import { store } from 'app/store/store';
@@ -59,9 +60,10 @@ import {
   supplementaryQueryTypes,
 } from '../utils/supplementaryQueries';
 
+import { saveCorrelationsAction } from './explorePane';
 import { addHistoryItem, historyUpdatedAction, loadRichHistory } from './history';
 import { updateTime } from './time';
-import { createCacheKey, filterLogRowsByIndex, getResultsFromCache } from './utils';
+import { createCacheKey, filterLogRowsByIndex, getDatasourceUIDs, getResultsFromCache } from './utils';
 
 /**
  * Derives from explore state if a given Explore pane is waiting for more data to be received
@@ -319,6 +321,7 @@ export const changeQueries = createAsyncThunk<void, ChangeQueriesPayload>(
   async ({ queries, exploreId }, { getState, dispatch }) => {
     let queriesImported = false;
     const oldQueries = getState().explore.panes[exploreId]!.queries;
+    const rootUID = getState().explore.panes[exploreId]!.datasourceInstance?.uid;
 
     for (const newQuery of queries) {
       for (const oldQuery of oldQueries) {
@@ -328,6 +331,16 @@ export const changeQueries = createAsyncThunk<void, ChangeQueriesPayload>(
           await dispatch(importQueries(exploreId, oldQueries, queryDatasource, targetDS, newQuery.refId));
           queriesImported = true;
         }
+
+        if (
+          rootUID === MIXED_DATASOURCE_NAME &&
+          newQuery.refId === oldQuery.refId &&
+          newQuery.datasource?.uid !== oldQuery.datasource?.uid
+        ) {
+          const datasourceUIDs = getDatasourceUIDs(MIXED_DATASOURCE_NAME, queries);
+          const correlations = await getCorrelationsBySourceUIDs(datasourceUIDs);
+          dispatch(saveCorrelationsAction({ exploreId: exploreId, correlations: correlations.correlations || [] }));
+        }
       }
     }
 
@@ -481,7 +494,7 @@ export const runQueries = createAsyncThunk<void, RunQueriesOptions>(
   async ({ exploreId, preserveCache }, { dispatch, getState }) => {
     dispatch(updateTime({ exploreId }));
 
-    const correlations$ = getCorrelations();
+    const correlations$ = getCorrelations(exploreId);
 
     // We always want to clear cache unless we explicitly pass preserveCache parameter
     if (preserveCache !== true) {
@@ -1134,15 +1147,15 @@ export const queryReducer = (state: ExploreItemState, action: AnyAction): Explor
 /**
  * Creates an observable that emits correlations once they are loaded
  */
-const getCorrelations = () => {
+const getCorrelations = (exploreId: string) => {
   return new Observable<CorrelationData[]>((subscriber) => {
-    const existingCorrelations = store.getState().explore.correlations;
+    const existingCorrelations = store.getState().explore.panes[exploreId]?.correlations;
     if (existingCorrelations) {
       subscriber.next(existingCorrelations);
       subscriber.complete();
     } else {
       const unsubscribe = store.subscribe(() => {
-        const { correlations } = store.getState().explore;
+        const correlations = store.getState().explore.panes[exploreId]?.correlations;
         if (correlations) {
           unsubscribe();
           subscriber.next(correlations);

public/app/features/explore/state/utils.ts

@@ -1,3 +1,5 @@
+import { uniq } from 'lodash';
+
 import {
   AbsoluteTimeRange,
   DataSourceApi,
@@ -15,7 +17,8 @@ import {
   isDateTime,
   toUtc,
 } from '@grafana/data';
-import { DataSourceRef, TimeZone } from '@grafana/schema';
+import { DataQuery, DataSourceRef, TimeZone } from '@grafana/schema';
+import { MIXED_DATASOURCE_NAME } from 'app/plugins/datasource/mixed/MixedDataSource';
 import { ExplorePanelData } from 'app/types';
 import { ExploreItemState } from 'app/types/explore';
 
@@ -67,6 +70,7 @@ export const makeExplorePaneState = (): ExploreItemState => ({
   richHistory: [],
   supplementaryQueries: loadSupplementaryQueries(),
   panelsState: {},
+  correlations: undefined,
 });
 
 export const createEmptyQueryResponse = (): ExplorePanelData => ({
@@ -205,3 +209,11 @@ export const filterLogRowsByIndex = (
 
   return logRows;
 };
+
+export const getDatasourceUIDs = (datasourceUID: string, queries: DataQuery[]): string[] => {
+  if (datasourceUID === MIXED_DATASOURCE_NAME) {
+    return uniq(queries.map((query) => query.datasource?.uid).filter((uid): uid is string => !!uid));
+  } else {
+    return [datasourceUID];
+  }
+};

public/app/types/explore.ts

@@ -33,8 +33,6 @@ export interface ExploreState {
 
   panes: Record<string, ExploreItemState | undefined>;
 
-  correlations?: CorrelationData[];
-
   /**
    * Settings for rich history (note: filters are stored per each pane separately)
    */
@@ -192,6 +190,8 @@ export interface ExploreItemState {
   supplementaryQueries: SupplementaryQueries;
 
   panelsState: ExplorePanelsState;
+
+  correlations?: CorrelationData[];
 }
 
 export interface ExploreUpdateState {