[Response Ops][Alerting] Refactor ExecutionHandler stage 1

x-pack/plugins/alerting/server/task_runner/action_scheduler/get_summarized_alerts.test.ts

@@ -0,0 +1,127 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { getSummarizedAlerts } from './get_summarized_alerts';
+import { alertsClientMock } from '../../alerts_client/alerts_client.mock';
+import { mockAAD } from '../fixtures';
+import { ALERT_UUID } from '@kbn/rule-data-utils';
+import { generateAlert } from './test_fixtures';
+import { getErrorSource } from '@kbn/task-manager-plugin/server/task_running';
+
+const alertsClient = alertsClientMock.create();
+
+describe('getSummarizedAlerts', () => {
+  const newAlert1 = generateAlert({ id: 1 });
+  const newAlert2 = generateAlert({ id: 2 });
+  const alerts = { ...newAlert1, ...newAlert2 };
+
+  beforeEach(() => {
+    jest.resetAllMocks();
+  });
+
+  test('should call alertsClient.getSummarizedAlerts with the correct params', async () => {
+    const summarizedAlerts = {
+      new: {
+        count: 2,
+        data: [
+          { ...mockAAD, [ALERT_UUID]: alerts[1].getUuid() },
+          { ...mockAAD, [ALERT_UUID]: alerts[2].getUuid() },
+        ],
+      },
+      ongoing: { count: 0, data: [] },
+      recovered: { count: 0, data: [] },
+    };
+    alertsClient.getSummarizedAlerts.mockResolvedValue(summarizedAlerts);
+    alertsClient.getProcessedAlerts.mockReturnValue(alerts);
+    const result = await getSummarizedAlerts({
+      alertsClient,
+      queryOptions: {
+        excludedAlertInstanceIds: [],
+        executionUuid: '123xyz',
+        ruleId: '1',
+        spaceId: 'test1',
+      },
+    });
+
+    expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
+      excludedAlertInstanceIds: [],
+      executionUuid: '123xyz',
+      ruleId: '1',
+      spaceId: 'test1',
+    });
+
+    expect(result).toEqual({
+      ...summarizedAlerts,
+      all: summarizedAlerts.new,
+    });
+  });
+
+  test('should throw error if alertsClient.getSummarizedAlerts throws error', async () => {
+    alertsClient.getSummarizedAlerts.mockImplementation(() => {
+      throw new Error('cannot get summarized alerts');
+    });
+
+    try {
+      await getSummarizedAlerts({
+        alertsClient,
+        queryOptions: {
+          excludedAlertInstanceIds: [],
+          executionUuid: '123xyz',
+          ruleId: '1',
+          spaceId: 'test1',
+        },
+      });
+    } catch (err) {
+      expect(getErrorSource(err)).toBe('framework');
+      expect(err.message).toBe('cannot get summarized alerts');
+    }
+  });
+
+  test('should remove alert from summarized alerts if it is new and has a maintenance window', async () => {
+    const newAlertWithMaintenanceWindow = generateAlert({
+      id: 1,
+      maintenanceWindowIds: ['mw-1'],
+    });
+    const alertsWithMaintenanceWindow = { ...newAlertWithMaintenanceWindow, ...newAlert2 };
+
+    const newAADAlerts = [
+      { ...mockAAD, [ALERT_UUID]: newAlertWithMaintenanceWindow[1].getUuid() },
+      { ...mockAAD, [ALERT_UUID]: alerts[2].getUuid() },
+    ];
+    const summarizedAlerts = {
+      new: { count: 2, data: newAADAlerts },
+      ongoing: { count: 0, data: [] },
+      recovered: { count: 0, data: [] },
+    };
+    alertsClient.getSummarizedAlerts.mockResolvedValue(summarizedAlerts);
+    alertsClient.getProcessedAlerts.mockReturnValue(alertsWithMaintenanceWindow);
+
+    const result = await getSummarizedAlerts({
+      alertsClient,
+      queryOptions: {
+        excludedAlertInstanceIds: [],
+        executionUuid: '123xyz',
+        ruleId: '1',
+        spaceId: 'test1',
+      },
+    });
+
+    expect(alertsClient.getSummarizedAlerts).toHaveBeenCalledWith({
+      excludedAlertInstanceIds: [],
+      executionUuid: '123xyz',
+      ruleId: '1',
+      spaceId: 'test1',
+    });
+
+    expect(result).toEqual({
+      new: { count: 1, data: [newAADAlerts[1]] },
+      ongoing: { count: 0, data: [] },
+      recovered: { count: 0, data: [] },
+      all: { count: 1, data: [newAADAlerts[1]] },
+    });
+  });
+});

x-pack/plugins/alerting/server/task_runner/action_scheduler/get_summarized_alerts.ts

@@ -0,0 +1,78 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { ALERT_UUID } from '@kbn/rule-data-utils';
+import { createTaskRunError, TaskErrorSource } from '@kbn/task-manager-plugin/server';
+import { GetSummarizedAlertsParams, IAlertsClient } from '../../alerts_client/types';
+import {
+  AlertInstanceContext,
+  AlertInstanceState,
+  CombinedSummarizedAlerts,
+  RuleAlertData,
+} from '../../types';
+
+interface GetSummarizedAlertsOpts<
+  State extends AlertInstanceState,
+  Context extends AlertInstanceContext,
+  ActionGroupIds extends string,
+  RecoveryActionGroupId extends string,
+  AlertData extends RuleAlertData
+> {
+  alertsClient: IAlertsClient<AlertData, State, Context, ActionGroupIds, RecoveryActionGroupId>;
+  queryOptions: GetSummarizedAlertsParams;
+}
+
+export const getSummarizedAlerts = async <
+  State extends AlertInstanceState,
+  Context extends AlertInstanceContext,
+  ActionGroupIds extends string,
+  RecoveryActionGroupId extends string,
+  AlertData extends RuleAlertData
+>({
+  alertsClient,
+  queryOptions,
+}: GetSummarizedAlertsOpts<
+  State,
+  Context,
+  ActionGroupIds,
+  RecoveryActionGroupId,
+  AlertData
+>): Promise<CombinedSummarizedAlerts> => {
+  let alerts;
+  try {
+    alerts = await alertsClient.getSummarizedAlerts!(queryOptions);
+  } catch (e) {
+    throw createTaskRunError(e, TaskErrorSource.FRAMEWORK);
+  }
+
+  /**
+   * We need to remove all new alerts with maintenance windows retrieved from
+   * getSummarizedAlerts because they might not have maintenance window IDs
+   * associated with them from maintenance windows with scoped query updated
+   * yet (the update call uses refresh: false). So we need to rely on the in
+   * memory alerts to do this.
+   */
+  const newAlertsInMemory = Object.values(alertsClient.getProcessedAlerts('new') || {}) || [];
+
+  const newAlertsWithMaintenanceWindowIds = newAlertsInMemory.reduce<string[]>((result, alert) => {
+    if (alert.getMaintenanceWindowIds().length > 0) {
+      result.push(alert.getUuid());
+    }
+    return result;
+  }, []);
+
+  const newAlerts = alerts.new.data.filter((alert) => {
+    return !newAlertsWithMaintenanceWindowIds.includes(alert[ALERT_UUID]);
+  });
+
+  const total = newAlerts.length + alerts.ongoing.count + alerts.recovered.count;
+  return {
+    ...alerts,
+    new: { count: newAlerts.length, data: newAlerts },
+    all: { count: total, data: [...newAlerts, ...alerts.ongoing.data, ...alerts.recovered.data] },
+  };
+};

x-pack/plugins/alerting/server/task_runner/action_scheduler/index.ts

@@ -0,0 +1,10 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+export { ActionScheduler } from './action_scheduler';
+export type { RunResult } from './action_scheduler';
+export type { RuleUrl } from './types';

x-pack/plugins/alerting/server/task_runner/rule_action_helper.test.ts → x-pack/plugins/alerting/server/task_runner/action_scheduler/rule_action_helper.test.ts

@@ -6,13 +6,12 @@
  */
 
 import { Logger } from '@kbn/logging';
-import { RuleAction } from '../types';
+import { RuleAction } from '../../types';
 import {
   generateActionHash,
   getSummaryActionsFromTaskState,
   isActionOnInterval,
   isSummaryAction,
-  isSummaryActionOnInterval,
   isSummaryActionThrottled,
   getSummaryActionTimeBounds,
 } from './rule_action_helper';
@@ -291,30 +290,6 @@ describe('rule_action_helper', () => {
     });
   });
 
-  describe('isSummaryActionOnInterval', () => {
-    test('returns true for a summary action on interval', () => {
-      expect(isSummaryActionOnInterval(mockSummaryAction)).toBe(true);
-    });
-
-    test('returns false for a non-summary ', () => {
-      expect(
-        isSummaryActionOnInterval({
-          ...mockAction,
-          frequency: { summary: false, notifyWhen: 'onThrottleInterval', throttle: '1h' },
-        })
-      ).toBe(false);
-    });
-
-    test('returns false for a summary per rule run ', () => {
-      expect(
-        isSummaryActionOnInterval({
-          ...mockAction,
-          frequency: { summary: true, notifyWhen: 'onActiveAlert', throttle: null },
-        })
-      ).toBe(false);
-    });
-  });
-
   describe('getSummaryActionTimeBounds', () => {
     test('returns undefined start and end action is not summary action', () => {
       expect(getSummaryActionTimeBounds(mockAction, { interval: '1m' }, null)).toEqual({

x-pack/plugins/alerting/server/task_runner/rule_action_helper.ts → x-pack/plugins/alerting/server/task_runner/action_scheduler/rule_action_helper.ts

@@ -12,7 +12,7 @@ import {
   RuleAction,
   RuleNotifyWhenTypeValues,
   ThrottledActions,
-} from '../../common';
+} from '../../../common';
 
 export const isSummaryAction = (action?: RuleAction) => {
   return action?.frequency?.summary ?? false;
@@ -28,10 +28,6 @@ export const isActionOnInterval = (action?: RuleAction) => {
   );
 };
 
-export const isSummaryActionOnInterval = (action: RuleAction) => {
-  return isActionOnInterval(action) && action.frequency?.summary;
-};
-
 export const isSummaryActionThrottled = ({
   action,
   throttledSummaryActions,
@@ -129,3 +125,25 @@ export const getSummaryActionTimeBounds = (
 
   return { start: startDate.valueOf(), end: now.valueOf() };
 };
+
+interface LogNumberOfFilteredAlertsOpts {
+  logger: Logger;
+  numberOfAlerts: number;
+  numberOfSummarizedAlerts: number;
+  action: RuleAction;
+}
+export const logNumberOfFilteredAlerts = ({
+  logger,
+  numberOfAlerts = 0,
+  numberOfSummarizedAlerts = 0,
+  action,
+}: LogNumberOfFilteredAlertsOpts) => {
+  const count = numberOfAlerts - numberOfSummarizedAlerts;
+  if (count > 0) {
+    logger.debug(
+      `(${count}) alert${count > 1 ? 's' : ''} ${
+        count > 1 ? 'have' : 'has'
+      } been filtered out for: ${action.actionTypeId}:${action.uuid}`
+    );
+  }
+};

x-pack/plugins/alerting/server/task_runner/action_scheduler/schedulers/index.ts

@@ -0,0 +1,10 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+export { SystemActionScheduler } from './system_action_scheduler';
+export { SummaryActionScheduler } from './summary_action_scheduler';
+export { PerAlertActionScheduler } from './per_alert_action_scheduler';