Alerting: Show annotations markers in TimeSeries panel when using Loki as … (#72084)

* WIP: Show annotations markers in TimeSeries panel when using Loki as alert state history

* WIP changes

* Fix converting log records to data frame for panel

* Move fetching alert state history with Loki to the PannelQueryRunner to keep the panel flow

* use dasboardUID and panelUID for requesting Loki ash

* fix wrong prettier change

* Only request loki ash when having alertstate

* Use panelID as param in history query

* Refactor: move getRuleHistoryRecordsForPanel and remove filtering code as is not used

* Adress PR review comments

* Add try catch for ash request

* Add tests for updatePanelDataWithASHFromLoki method

* Address PR review suggestions

* review suggestion

* Add test for logRecordsToDataFrameForPanel method

* pr Review nit suggestion

* Dont show toast messages from Loki request

public/app/features/alerting/state/alertDef.ts

@@ -1,7 +1,7 @@
 import { isArray, reduce } from 'lodash';
 
 import { IconName } from '@grafana/ui';
-import { QueryPartDef, QueryPart } from 'app/features/alerting/state/query_part';
+import { QueryPart, QueryPartDef } from 'app/features/alerting/state/query_part';
 
 const alertQueryDef = new QueryPartDef({
   type: 'query',
@@ -84,7 +84,7 @@ function createReducerPart(model: any) {
 
 // state can also contain a "Reason", ie. "Alerting (NoData)" which indicates that the actual state is "Alerting" but
 // the reason it is set to "Alerting" is "NoData"; a lack of data points to evaluate.
-function normalizeAlertState(state: string) {
+export function normalizeAlertState(state: string) {
   return state.toLowerCase().replace(/_/g, '').split(' ')[0];
 }
 

public/app/features/alerting/unified/Analytics.ts

@@ -16,6 +16,7 @@ export const LogMessages = {
   cancelSavingAlertRule: 'user canceled alert rule creation',
   successSavingAlertRule: 'alert rule saved successfully',
   unknownMessageFromError: 'unknown messageFromError',
+  errorGettingLokiHistory: 'error getting Loki history',
 };
 
 // logInfo from '@grafana/runtime' should be used, but it doesn't handle Grafana JS Agent correctly

public/app/features/alerting/unified/components/rules/state-history/common.test.ts

@@ -1,4 +1,23 @@
-import { extractCommonLabels, Label, omitLabels } from './common';
+import { rest } from 'msw';
+import { setupServer } from 'msw/node';
+
+import {
+  AlertState,
+  DataFrameJSON,
+  FieldType,
+  getDefaultTimeRange,
+  LoadingState,
+  PanelData,
+  toDataFrame,
+} from '@grafana/data';
+import { setBackendSrv } from '@grafana/runtime';
+import { backendSrv } from 'app/core/services/backend_srv';
+
+import 'whatwg-fetch';
+import { StateHistoryImplementation } from '../../../hooks/useStateHistoryModal';
+
+import * as common from './common';
+import { extractCommonLabels, Label, omitLabels, updatePanelDataWithASHFromLoki } from './common';
 
 test('extractCommonLabels', () => {
   const labels: Label[][] = [
@@ -48,3 +67,120 @@ test('omitLabels with no common labels', () => {
 
   expect(omitLabels(labels, commonLabels)).toStrictEqual(labels);
 });
+
+const server = setupServer();
+
+beforeAll(() => {
+  setBackendSrv(backendSrv);
+  server.listen({ onUnhandledRequest: 'error' });
+
+  server.use(
+    rest.get('/api/v1/rules/history', (req, res, ctx) =>
+      res(
+        ctx.json<DataFrameJSON>({
+          data: {
+            values: [
+              [1681739580000, 1681739580000, 1681739580000],
+              [
+                {
+                  previous: 'Normal',
+                  current: 'Pending',
+                  values: {
+                    B: 0.010344684900897919,
+                    C: 1,
+                  },
+                  labels: {
+                    handler: '/api/prometheus/grafana/api/v1/rules',
+                  },
+                },
+                {
+                  previous: 'Normal',
+                  current: 'Pending',
+                  values: {
+                    B: 0.010344684900897919,
+                    C: 1,
+                  },
+                  dashboardUID: '',
+                  panelID: 0,
+                  labels: {
+                    handler: '/api/live/ws',
+                  },
+                },
+                {
+                  previous: 'Normal',
+                  current: 'Pending',
+                  values: {
+                    B: 0.010344684900897919,
+                    C: 1,
+                  },
+                  labels: {
+                    handler: '/api/folders/:uid/',
+                  },
+                },
+              ],
+            ],
+          },
+        })
+      )
+    )
+  );
+});
+
+afterAll(() => {
+  server.close();
+});
+
+jest.spyOn(common, 'getHistoryImplementation').mockImplementation(() => StateHistoryImplementation.Loki);
+const getHistoryImplementationMock = common.getHistoryImplementation as jest.MockedFunction<
+  typeof common.getHistoryImplementation
+>;
+const timeRange = getDefaultTimeRange();
+const panelDataProcessed: PanelData = {
+  alertState: {
+    id: 1,
+    dashboardId: 1,
+    panelId: 1,
+    state: AlertState.Alerting,
+  },
+  series: [
+    toDataFrame({
+      fields: [
+        { name: 'time', type: FieldType.time },
+        { name: 'score', type: FieldType.number },
+      ],
+    }),
+  ],
+  annotations: [toDataFrame([{ id: 'panelData' }]), toDataFrame([{ id: 'dashData' }])],
+  state: LoadingState.Done,
+  timeRange,
+};
+
+describe('updatePanelDataWithASHFromLoki', () => {
+  it('should return the same panelData if not using Loki as implementation', async () => {
+    getHistoryImplementationMock.mockImplementation(() => StateHistoryImplementation.Annotations);
+
+    const panelData = await updatePanelDataWithASHFromLoki(panelDataProcessed);
+
+    expect(panelData).toStrictEqual(panelDataProcessed);
+    expect(panelData.annotations).toHaveLength(2);
+  });
+
+  it('should return the correct panelData if using Loki as implementation', async () => {
+    getHistoryImplementationMock.mockImplementation(() => StateHistoryImplementation.Loki);
+
+    const panelData = await updatePanelDataWithASHFromLoki(panelDataProcessed);
+
+    expect(panelData.annotations).toHaveLength(5);
+  });
+
+  it('should return the same panelData if Loki call throws an error', async () => {
+    getHistoryImplementationMock.mockImplementation(() => StateHistoryImplementation.Loki);
+
+    server.use(rest.get('/api/v1/rules/history', (req, res, ctx) => res(ctx.status(500))));
+
+    const panelData = await updatePanelDataWithASHFromLoki(panelDataProcessed);
+
+    expect(panelData).toStrictEqual(panelDataProcessed);
+    expect(panelData.annotations).toHaveLength(2);
+  });
+});

public/app/features/alerting/unified/components/rules/state-history/common.ts

@@ -1,7 +1,15 @@
-import { isEqual, uniqBy } from 'lodash';
+import { cloneDeep, groupBy, isEqual, uniqBy } from 'lodash';
+import { lastValueFrom } from 'rxjs';
 
+import { DataFrame, DataFrameJSON, PanelData } from '@grafana/data';
+import { config, getBackendSrv } from '@grafana/runtime';
 import { GrafanaAlertStateWithReason } from 'app/types/unified-alerting-dto';
 
+import { logInfo, LogMessages } from '../../../Analytics';
+import { StateHistoryImplementation } from '../../../hooks/useStateHistoryModal';
+
+import { isLine, isNumbers, logRecordsToDataFrameForPanel } from './useRuleHistoryRecords';
+
 export interface Line {
   previous: GrafanaAlertStateWithReason;
   current: GrafanaAlertStateWithReason;
@@ -37,3 +45,108 @@ export function extractCommonLabels(labels: Label[][]): Label[] {
 
   return commonLabels;
 }
+
+export const getLogRecordsByInstances = (stateHistory?: DataFrameJSON) => {
+  // merge timestamp with "line"
+  const tsValues = stateHistory?.data?.values[0] ?? [];
+  const timestamps: number[] = isNumbers(tsValues) ? tsValues : [];
+  const lines = stateHistory?.data?.values[1] ?? [];
+
+  const logRecords = timestamps.reduce((acc: LogRecord[], timestamp: number, index: number) => {
+    const line = lines[index];
+    // values property can be undefined for some instance states (e.g. NoData)
+    if (isLine(line)) {
+      acc.push({ timestamp, line });
+    }
+
+    return acc;
+  }, []);
+
+  // group all records by alert instance (unique set of labels)
+  const logRecordsByInstance = groupBy(logRecords, (record: LogRecord) => {
+    return JSON.stringify(record.line.labels);
+  });
+
+  return { logRecordsByInstance, logRecords };
+};
+
+export function getRuleHistoryRecordsForPanel(stateHistory?: DataFrameJSON) {
+  if (!stateHistory) {
+    return { dataFrames: [] };
+  }
+  const theme = config.theme2;
+
+  const { logRecordsByInstance } = getLogRecordsByInstances(stateHistory);
+
+  const groupedLines = Object.entries(logRecordsByInstance);
+
+  const dataFrames: DataFrame[] = groupedLines.map<DataFrame>(([key, records]) => {
+    return logRecordsToDataFrameForPanel(key, records, theme);
+  });
+
+  return {
+    dataFrames,
+  };
+}
+
+export const getHistoryImplementation = () => {
+  // can be "loki", "multiple" or "annotations"
+  const stateHistoryBackend = config.unifiedAlerting.alertStateHistoryBackend;
+  // can be "loki" or "annotations"
+  const stateHistoryPrimary = config.unifiedAlerting.alertStateHistoryPrimary;
+
+  // if "loki" is either the backend or the primary, show the new state history implementation
+  const usingNewAlertStateHistory = [stateHistoryBackend, stateHistoryPrimary].some(
+    (implementation) => implementation === StateHistoryImplementation.Loki
+  );
+  const implementation = usingNewAlertStateHistory
+    ? StateHistoryImplementation.Loki
+    : StateHistoryImplementation.Annotations;
+  return implementation;
+};
+
+export const updatePanelDataWithASHFromLoki = async (panelDataProcessed: PanelData) => {
+  //--- check if alert state history uses Loki as implementation, if so, fetch data from Loki state history and concat it to annotations
+  const historyImplementation = getHistoryImplementation();
+  const usingLokiAsImplementation = historyImplementation === StateHistoryImplementation.Loki;
+
+  const notShouldFetchLokiAsh =
+    !usingLokiAsImplementation ||
+    !panelDataProcessed.alertState?.dashboardId ||
+    !panelDataProcessed.alertState?.panelId;
+
+  if (notShouldFetchLokiAsh) {
+    return panelDataProcessed;
+  }
+
+  try {
+    // fetch data from Loki state history
+    let annotationsWithHistory = await lastValueFrom(
+      getBackendSrv().fetch<DataFrameJSON>({
+        url: '/api/v1/rules/history',
+        method: 'GET',
+        params: {
+          panelID: panelDataProcessed.request?.panelId,
+          dashboardUID: panelDataProcessed.request?.dashboardUID,
+          from: panelDataProcessed.timeRange.from.unix(),
+          to: panelDataProcessed.timeRange.to.unix(),
+          limit: 250,
+        },
+        showErrorAlert: false,
+        showSuccessAlert: false,
+      })
+    );
+    const records = getRuleHistoryRecordsForPanel(annotationsWithHistory.data);
+    const clonedPanel = cloneDeep(panelDataProcessed);
+    // annotations can be undefined
+    clonedPanel.annotations = panelDataProcessed.annotations
+      ? panelDataProcessed.annotations.concat(records.dataFrames)
+      : panelDataProcessed.annotations;
+    return clonedPanel;
+  } catch (error) {
+    logInfo(LogMessages.errorGettingLokiHistory, {
+      error: error instanceof Error ? error.message : 'Unknown error getting Loki ash',
+    });
+    return panelDataProcessed;
+  }
+};

public/app/features/alerting/unified/components/rules/state-history/useRuleHistoryRecords.test.tsx

@@ -1,11 +1,11 @@
 import { createTheme, FieldType } from '@grafana/data';
 
 import { LogRecord } from './common';
-import { logRecordsToDataFrame } from './useRuleHistoryRecords';
+import { logRecordsToDataFrame, logRecordsToDataFrameForPanel } from './useRuleHistoryRecords';
+
+const theme = createTheme();
 
 describe('logRecordsToDataFrame', () => {
-  const theme = createTheme();
-
   it('should convert instance history records into a data frame', () => {
     const instanceLabels = { foo: 'bar', severity: 'critical', cluster: 'dev-us' };
     const records: LogRecord[] = [
@@ -102,3 +102,67 @@ describe('logRecordsToDataFrame', () => {
     expect(frame.fields[1].config.displayName).toBe('severity=critical');
   });
 });
+
+describe('logRecordsToDataFrameForPanel', () => {
+  it('should return correct data frame records', () => {
+    const instanceLabels = { foo: 'bar', severity: 'critical', cluster: 'dev-us' };
+    const records: LogRecord[] = [
+      {
+        timestamp: 1000000,
+        line: { previous: 'Normal', current: 'Alerting', labels: instanceLabels, values: { A: 10, B: 90 } },
+      },
+      {
+        timestamp: 1000050,
+        line: { previous: 'Alerting', current: 'Normal', labels: instanceLabels },
+      },
+    ];
+
+    const frame = logRecordsToDataFrameForPanel(JSON.stringify(instanceLabels), records, theme);
+
+    expect(frame.fields).toHaveLength(6);
+    expect(frame).toHaveLength(2);
+    expect(frame.fields[0]).toMatchObject({
+      name: 'time',
+      type: FieldType.time,
+      values: [1000000, 1000050],
+    });
+    expect(frame.fields[1]).toMatchObject({
+      name: 'alertId',
+      type: FieldType.string,
+      values: [1, 1],
+    });
+    expect(frame.fields[2]).toMatchObject({
+      name: 'newState',
+      type: FieldType.string,
+      values: ['Alerting', 'Normal'],
+    });
+    expect(frame.fields[3]).toMatchObject({
+      name: 'prevState',
+      type: FieldType.string,
+      values: ['Normal', 'Alerting'],
+    });
+    expect(frame.fields[4]).toMatchObject({
+      name: 'color',
+      type: FieldType.string,
+      values: [theme.colors.error.main, theme.colors.success.main],
+    });
+    expect(frame.fields[5]).toMatchObject({
+      name: 'data',
+      type: FieldType.other,
+      values: [
+        [
+          { metric: 'foo', value: 'bar' },
+          { metric: 'severity', value: 'critical' },
+          { metric: 'cluster', value: 'dev-us' },
+          { metric: ' Values', value: '{A= 10, B= 90}' },
+        ],
+        [
+          { metric: 'foo', value: 'bar' },
+          { metric: 'severity', value: 'critical' },
+          { metric: 'cluster', value: 'dev-us' },
+          { metric: '', value: '' },
+        ],
+      ],
+    });
+  });
+});

public/app/features/alerting/unified/components/rules/state-history/useRuleHistoryRecords.tsx

@@ -1,4 +1,3 @@
-import { groupBy } from 'lodash';
 import { useMemo } from 'react';
 
 import {
@@ -12,34 +11,17 @@ import {
 import { fieldIndexComparer } from '@grafana/data/src/field/fieldComparers';
 import { MappingType, ThresholdsMode } from '@grafana/schema';
 import { useTheme2 } from '@grafana/ui';
+import { normalizeAlertState } from 'app/features/alerting/state/alertDef';
 
 import { labelsMatchMatchers, parseMatchers } from '../../../utils/alertmanager';
 
-import { extractCommonLabels, Line, LogRecord, omitLabels } from './common';
+import { extractCommonLabels, getLogRecordsByInstances, Line, LogRecord, omitLabels } from './common';
 
 export function useRuleHistoryRecords(stateHistory?: DataFrameJSON, filter?: string) {
   const theme = useTheme2();
 
   return useMemo(() => {
-    // merge timestamp with "line"
+    const { logRecordsByInstance, logRecords } = getLogRecordsByInstances(stateHistory);
-    const tsValues = stateHistory?.data?.values[0] ?? [];
-    const timestamps: number[] = isNumbers(tsValues) ? tsValues : [];
-    const lines = stateHistory?.data?.values[1] ?? [];
-
-    const logRecords = timestamps.reduce((acc: LogRecord[], timestamp: number, index: number) => {
-      const line = lines[index];
-      // values property can be undefined for some instance states (e.g. NoData)
-      if (isLine(line)) {
-        acc.push({ timestamp, line });
-      }
-
-      return acc;
-    }, []);
-
-    // group all records by alert instance (unique set of labels)
-    const logRecordsByInstance = groupBy(logRecords, (record: LogRecord) => {
-      return JSON.stringify(record.line.labels);
-    });
 
     // CommonLabels should not be affected by the filter
     // find common labels so we can extract those from the instances
@@ -153,3 +135,132 @@ export function logRecordsToDataFrame(
 
   return frame;
 }
+
+interface MetricValuePair {
+  metric: string;
+  value: string;
+}
+
+function logRecordToData(record: LogRecord) {
+  let labelsInLogs: MetricValuePair[] = [];
+  let valuesInLogs: MetricValuePair = { metric: '', value: '' };
+  if (record.line.labels) {
+    const { labels } = record.line;
+    const labelsArray = Object.entries(labels);
+    labelsInLogs = labelsArray.map(([key, value]) => ({ metric: key, value }));
+  }
+
+  let values = record.line.values;
+  if (values) {
+    const valuesArray = Object.entries(values);
+    const valuesData = valuesArray.map(([key, value]) => ({ metric: key, value: value.toString() }));
+    //convert valuesInloGS to a one Data entry
+    valuesInLogs = valuesData.reduce<MetricValuePair>(
+      (acc, cur) => {
+        acc.value = acc.value.length > 0 ? acc.value + ', ' : acc.value;
+        acc.value = cur.metric.length > 0 ? acc.value + cur.metric + '= ' + cur.value : acc.value;
+        return acc;
+      },
+      { metric: ' Values', value: '' }
+    );
+    if (valuesInLogs.value.length > 0) {
+      valuesInLogs.value = '{' + valuesInLogs.value + '}';
+      return [...labelsInLogs, valuesInLogs];
+    } else {
+      return labelsInLogs;
+    }
+  }
+  return [...labelsInLogs, valuesInLogs];
+}
+
+// Convert log records to data frame for panel
+export function logRecordsToDataFrameForPanel(
+  instanceLabels: string,
+  records: LogRecord[],
+  theme: GrafanaTheme2
+): DataFrame {
+  const timeField: DataFrameField = {
+    name: 'time',
+    type: FieldType.time,
+    values: records.map((record) => record.timestamp),
+    config: { displayName: 'Time', custom: { fillOpacity: 100 } },
+  };
+
+  const timeIndex = timeField.values.map((_, index) => index);
+  timeIndex.sort(fieldIndexComparer(timeField));
+
+  const frame: DataFrame = {
+    fields: [
+      {
+        ...timeField,
+        values: timeField.values.map((_, i) => timeField.values[timeIndex[i]]),
+      },
+      {
+        name: 'alertId',
+        type: FieldType.string,
+        values: records.map((_) => 1),
+        config: {
+          displayName: 'AlertId',
+          custom: { fillOpacity: 100 },
+        },
+      },
+      {
+        name: 'newState',
+        type: FieldType.string,
+        values: records.map((record) => record.line.current),
+        config: {
+          displayName: 'newState',
+          custom: { fillOpacity: 100 },
+        },
+      },
+      {
+        name: 'prevState',
+        type: FieldType.string,
+        values: records.map((record) => record.line.previous),
+        config: {
+          displayName: 'prevState',
+          custom: { fillOpacity: 100 },
+        },
+      },
+      {
+        name: 'color',
+        type: FieldType.string,
+        values: records.map((record) => {
+          const normalizedState = normalizeAlertState(record.line.current);
+          switch (normalizedState) {
+            case 'firing':
+            case 'alerting':
+            case 'error':
+              return theme.colors.error.main;
+            case 'pending':
+              return theme.colors.warning.main;
+            case 'normal':
+              return theme.colors.success.main;
+            case 'nodata':
+              return theme.colors.info.main;
+            case 'paused':
+              return theme.colors.text.disabled;
+            default:
+              return theme.colors.info.main;
+          }
+        }),
+        config: {},
+      },
+      {
+        name: 'data',
+        type: FieldType.other,
+        values: records.map((record) => {
+          return logRecordToData(record);
+        }),
+        config: {},
+      },
+    ],
+    length: timeField.values.length,
+    name: instanceLabels,
+  };
+
+  frame.fields.forEach((field) => {
+    field.display = getDisplayProcessor({ field, theme });
+  });
+  return frame;
+}

public/app/features/alerting/unified/hooks/useStateHistoryModal.tsx

@@ -2,14 +2,15 @@ import { css } from '@emotion/css';
 import React, { lazy, Suspense, useCallback, useMemo, useState } from 'react';
 
 import { GrafanaTheme2 } from '@grafana/data';
-import { config } from '@grafana/runtime';
 import { Modal, useStyles2 } from '@grafana/ui';
 import { RulerGrafanaRuleDTO } from 'app/types/unified-alerting-dto';
 
+import { getHistoryImplementation } from '../components/rules/state-history/common';
+
 const AnnotationsStateHistory = lazy(() => import('../components/rules/state-history/StateHistory'));
 const LokiStateHistory = lazy(() => import('../components/rules/state-history/LokiStateHistory'));
 
-enum StateHistoryImplementation {
+export enum StateHistoryImplementation {
   Loki = 'loki',
   Annotations = 'annotations',
 }
@@ -20,18 +21,7 @@ function useStateHistoryModal() {
 
   const styles = useStyles2(getStyles);
 
-  // can be "loki", "multiple" or "annotations"
+  const implementation = getHistoryImplementation();
-  const stateHistoryBackend = config.unifiedAlerting.alertStateHistoryBackend;
-  // can be "loki" or "annotations"
-  const stateHistoryPrimary = config.unifiedAlerting.alertStateHistoryPrimary;
-
-  // if "loki" is either the backend or the primary, show the new state history implementation
-  const usingNewAlertStateHistory = [stateHistoryBackend, stateHistoryPrimary].some(
-    (implementation) => implementation === StateHistoryImplementation.Loki
-  );
-  const implementation = usingNewAlertStateHistory
-    ? StateHistoryImplementation.Loki
-    : StateHistoryImplementation.Annotations;
 
   const dismissModal = useCallback(() => {
     setRule(undefined);

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

@@ -3,6 +3,7 @@ import { Observable, of, ReplaySubject, Unsubscribable } from 'rxjs';
 import { map, mergeMap } from 'rxjs/operators';
 
 import {
+  ApplyFieldOverrideOptions,
   applyFieldOverrides,
   compareArrayValues,
   compareDataFrameStructures,
@@ -19,18 +20,18 @@ import {
   getDefaultTimeRange,
   LoadingState,
   PanelData,
+  preProcessPanelData,
   rangeUtil,
   ScopedVars,
+  StreamingDataFrame,
   TimeRange,
   TimeZone,
   toDataFrame,
   transformDataFrame,
-  preProcessPanelData,
-  ApplyFieldOverrideOptions,
-  StreamingDataFrame,
 } from '@grafana/data';
 import { getTemplateSrv, toDataQueryError } from '@grafana/runtime';
 import { ExpressionDatasourceRef } from '@grafana/runtime/src/utils/DataSourceWithBackend';
+import { updatePanelDataWithASHFromLoki } from 'app/features/alerting/unified/components/rules/state-history/common';
 import { isStreamingDataFrame } from 'app/features/live/data/utils';
 import { getDatasourceSrv } from 'app/features/plugins/datasource_srv';
 
@@ -324,8 +325,11 @@ export class PanelQueryRunner {
     }
 
     this.subscription = panelData.subscribe({
-      next: (data) => {
+      next: async (data) => {
-        this.lastResult = skipPreProcess ? data : preProcessPanelData(data, this.lastResult);
+        this.lastResult = skipPreProcess
+          ? data
+          : await updatePanelDataWithASHFromLoki(preProcessPanelData(data, this.lastResult));
+
         // Store preprocessed query results for applying overrides later on in the pipeline
         this.subject.next(this.lastResult);
       },