[Metrics Alerts] Add functional and unit tests (#60442) (#60774)

* Add tests for metric threshold alerts

* Fix count aggregator

* Remove redundant typedefs

Co-authored-by: Elastic Machine <elasticmachine@users.noreply.github.com>

Co-authored-by: Elastic Machine <elasticmachine@users.noreply.github.com>

x-pack/plugins/infra/server/lib/alerting/metric_threshold/metric_threshold_executor.test.ts

@@ -0,0 +1,244 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import { createMetricThresholdExecutor, FIRED_ACTIONS } from './metric_threshold_executor';
+import { Comparator, AlertStates } from './types';
+import * as mocks from './test_mocks';
+import { AlertExecutorOptions } from '../../../../../alerting/server';
+
+const executor = createMetricThresholdExecutor('test') as (opts: {
+  params: AlertExecutorOptions['params'];
+  services: { callCluster: AlertExecutorOptions['params']['callCluster'] };
+}) => Promise<void>;
+const alertInstances = new Map();
+
+const services = {
+  callCluster(_: string, { body }: any) {
+    const metric = body.query.bool.filter[1].exists.field;
+    if (body.aggs.groupings) {
+      if (body.aggs.groupings.composite.after) {
+        return mocks.compositeEndResponse;
+      }
+      if (metric === 'test.metric.2') {
+        return mocks.alternateCompositeResponse;
+      }
+      return mocks.basicCompositeResponse;
+    }
+    if (metric === 'test.metric.2') {
+      return mocks.alternateMetricResponse;
+    }
+    return mocks.basicMetricResponse;
+  },
+  alertInstanceFactory(instanceID: string) {
+    let state: any;
+    const actionQueue: any[] = [];
+    const instance = {
+      actionQueue: [],
+      get state() {
+        return state;
+      },
+      get mostRecentAction() {
+        return actionQueue.pop();
+      },
+    };
+    alertInstances.set(instanceID, instance);
+    return {
+      instanceID,
+      scheduleActions(id: string, action: any) {
+        actionQueue.push({ id, action });
+      },
+      replaceState(newState: any) {
+        state = newState;
+      },
+    };
+  },
+};
+
+const baseCriterion = {
+  aggType: 'avg',
+  metric: 'test.metric.1',
+  timeSize: 1,
+  timeUnit: 'm',
+  indexPattern: 'metricbeat-*',
+};
+describe('The metric threshold alert type', () => {
+  describe('querying the entire infrastructure', () => {
+    const instanceID = 'test-*';
+    const execute = (comparator: Comparator, threshold: number[]) =>
+      executor({
+        services,
+        params: {
+          criteria: [
+            {
+              ...baseCriterion,
+              comparator,
+              threshold,
+            },
+          ],
+        },
+      });
+    test('alerts as expected with the > comparator', async () => {
+      await execute(Comparator.GT, [0.75]);
+      expect(alertInstances.get(instanceID).mostRecentAction.id).toBe(FIRED_ACTIONS.id);
+      expect(alertInstances.get(instanceID).state.alertState).toBe(AlertStates.ALERT);
+      await execute(Comparator.GT, [1.5]);
+      expect(alertInstances.get(instanceID).mostRecentAction).toBe(undefined);
+      expect(alertInstances.get(instanceID).state.alertState).toBe(AlertStates.OK);
+    });
+    test('alerts as expected with the < comparator', async () => {
+      await execute(Comparator.LT, [1.5]);
+      expect(alertInstances.get(instanceID).mostRecentAction.id).toBe(FIRED_ACTIONS.id);
+      expect(alertInstances.get(instanceID).state.alertState).toBe(AlertStates.ALERT);
+      await execute(Comparator.LT, [0.75]);
+      expect(alertInstances.get(instanceID).mostRecentAction).toBe(undefined);
+      expect(alertInstances.get(instanceID).state.alertState).toBe(AlertStates.OK);
+    });
+    test('alerts as expected with the >= comparator', async () => {
+      await execute(Comparator.GT_OR_EQ, [0.75]);
+      expect(alertInstances.get(instanceID).mostRecentAction.id).toBe(FIRED_ACTIONS.id);
+      expect(alertInstances.get(instanceID).state.alertState).toBe(AlertStates.ALERT);
+      await execute(Comparator.GT_OR_EQ, [1.0]);
+      expect(alertInstances.get(instanceID).mostRecentAction.id).toBe(FIRED_ACTIONS.id);
+      expect(alertInstances.get(instanceID).state.alertState).toBe(AlertStates.ALERT);
+      await execute(Comparator.GT_OR_EQ, [1.5]);
+      expect(alertInstances.get(instanceID).mostRecentAction).toBe(undefined);
+      expect(alertInstances.get(instanceID).state.alertState).toBe(AlertStates.OK);
+    });
+    test('alerts as expected with the <= comparator', async () => {
+      await execute(Comparator.LT_OR_EQ, [1.5]);
+      expect(alertInstances.get(instanceID).mostRecentAction.id).toBe(FIRED_ACTIONS.id);
+      expect(alertInstances.get(instanceID).state.alertState).toBe(AlertStates.ALERT);
+      await execute(Comparator.LT_OR_EQ, [1.0]);
+      expect(alertInstances.get(instanceID).mostRecentAction.id).toBe(FIRED_ACTIONS.id);
+      expect(alertInstances.get(instanceID).state.alertState).toBe(AlertStates.ALERT);
+      await execute(Comparator.LT_OR_EQ, [0.75]);
+      expect(alertInstances.get(instanceID).mostRecentAction).toBe(undefined);
+      expect(alertInstances.get(instanceID).state.alertState).toBe(AlertStates.OK);
+    });
+    test('alerts as expected with the between comparator', async () => {
+      await execute(Comparator.BETWEEN, [0, 1.5]);
+      expect(alertInstances.get(instanceID).mostRecentAction.id).toBe(FIRED_ACTIONS.id);
+      expect(alertInstances.get(instanceID).state.alertState).toBe(AlertStates.ALERT);
+      await execute(Comparator.BETWEEN, [0, 0.75]);
+      expect(alertInstances.get(instanceID).mostRecentAction).toBe(undefined);
+      expect(alertInstances.get(instanceID).state.alertState).toBe(AlertStates.OK);
+    });
+  });
+
+  describe('querying with a groupBy parameter', () => {
+    const execute = (comparator: Comparator, threshold: number[]) =>
+      executor({
+        services,
+        params: {
+          groupBy: 'something',
+          criteria: [
+            {
+              ...baseCriterion,
+              comparator,
+              threshold,
+            },
+          ],
+        },
+      });
+    const instanceIdA = 'test-a';
+    const instanceIdB = 'test-b';
+    test('sends an alert when all groups pass the threshold', async () => {
+      await execute(Comparator.GT, [0.75]);
+      expect(alertInstances.get(instanceIdA).mostRecentAction.id).toBe(FIRED_ACTIONS.id);
+      expect(alertInstances.get(instanceIdA).state.alertState).toBe(AlertStates.ALERT);
+      expect(alertInstances.get(instanceIdB).mostRecentAction.id).toBe(FIRED_ACTIONS.id);
+      expect(alertInstances.get(instanceIdB).state.alertState).toBe(AlertStates.ALERT);
+    });
+    test('sends an alert when only some groups pass the threshold', async () => {
+      await execute(Comparator.LT, [1.5]);
+      expect(alertInstances.get(instanceIdA).mostRecentAction.id).toBe(FIRED_ACTIONS.id);
+      expect(alertInstances.get(instanceIdA).state.alertState).toBe(AlertStates.ALERT);
+      expect(alertInstances.get(instanceIdB).mostRecentAction).toBe(undefined);
+      expect(alertInstances.get(instanceIdB).state.alertState).toBe(AlertStates.OK);
+    });
+    test('sends no alert when no groups pass the threshold', async () => {
+      await execute(Comparator.GT, [5]);
+      expect(alertInstances.get(instanceIdA).mostRecentAction).toBe(undefined);
+      expect(alertInstances.get(instanceIdA).state.alertState).toBe(AlertStates.OK);
+      expect(alertInstances.get(instanceIdB).mostRecentAction).toBe(undefined);
+      expect(alertInstances.get(instanceIdB).state.alertState).toBe(AlertStates.OK);
+    });
+  });
+
+  describe('querying with multiple criteria', () => {
+    const execute = (
+      comparator: Comparator,
+      thresholdA: number[],
+      thresholdB: number[],
+      groupBy: string = ''
+    ) =>
+      executor({
+        services,
+        params: {
+          groupBy,
+          criteria: [
+            {
+              ...baseCriterion,
+              comparator,
+              threshold: thresholdA,
+            },
+            {
+              ...baseCriterion,
+              comparator,
+              threshold: thresholdB,
+              metric: 'test.metric.2',
+            },
+          ],
+        },
+      });
+    test('sends an alert when all criteria cross the threshold', async () => {
+      const instanceID = 'test-*';
+      await execute(Comparator.GT_OR_EQ, [1.0], [3.0]);
+      expect(alertInstances.get(instanceID).mostRecentAction.id).toBe(FIRED_ACTIONS.id);
+      expect(alertInstances.get(instanceID).state.alertState).toBe(AlertStates.ALERT);
+    });
+    test('sends no alert when some, but not all, criteria cross the threshold', async () => {
+      const instanceID = 'test-*';
+      await execute(Comparator.LT_OR_EQ, [1.0], [3.0]);
+      expect(alertInstances.get(instanceID).mostRecentAction).toBe(undefined);
+      expect(alertInstances.get(instanceID).state.alertState).toBe(AlertStates.OK);
+    });
+    test('alerts only on groups that meet all criteria when querying with a groupBy parameter', async () => {
+      const instanceIdA = 'test-a';
+      const instanceIdB = 'test-b';
+      await execute(Comparator.GT_OR_EQ, [1.0], [3.0], 'something');
+      expect(alertInstances.get(instanceIdA).mostRecentAction.id).toBe(FIRED_ACTIONS.id);
+      expect(alertInstances.get(instanceIdA).state.alertState).toBe(AlertStates.ALERT);
+      expect(alertInstances.get(instanceIdB).mostRecentAction).toBe(undefined);
+      expect(alertInstances.get(instanceIdB).state.alertState).toBe(AlertStates.OK);
+    });
+  });
+  describe('querying with the count aggregator', () => {
+    const instanceID = 'test-*';
+    const execute = (comparator: Comparator, threshold: number[]) =>
+      executor({
+        services,
+        params: {
+          criteria: [
+            {
+              ...baseCriterion,
+              comparator,
+              threshold,
+              aggType: 'count',
+            },
+          ],
+        },
+      });
+    test('alerts based on the doc_count value instead of the aggregatedValue', async () => {
+      await execute(Comparator.GT, [2]);
+      expect(alertInstances.get(instanceID).mostRecentAction.id).toBe(FIRED_ACTIONS.id);
+      expect(alertInstances.get(instanceID).state.alertState).toBe(AlertStates.ALERT);
+      await execute(Comparator.LT, [1.5]);
+      expect(alertInstances.get(instanceID).mostRecentAction).toBe(undefined);
+      expect(alertInstances.get(instanceID).state.alertState).toBe(AlertStates.OK);
+    });
+  });
+});

x-pack/plugins/infra/server/lib/alerting/metric_threshold/metric_threshold_executor.ts

@@ -0,0 +1,255 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+import { mapValues } from 'lodash';
+import { i18n } from '@kbn/i18n';
+import { InfraDatabaseSearchResponse } from '../../adapters/framework/adapter_types';
+import { createAfterKeyHandler } from '../../../utils/create_afterkey_handler';
+import { getAllCompositeData } from '../../../utils/get_all_composite_data';
+import { networkTraffic } from '../../../../common/inventory_models/shared/metrics/snapshot/network_traffic';
+import { MetricExpressionParams, Comparator, AlertStates } from './types';
+import { AlertServices, AlertExecutorOptions } from '../../../../../alerting/server';
+
+interface Aggregation {
+  aggregatedIntervals: {
+    buckets: Array<{ aggregatedValue: { value: number }; doc_count: number }>;
+  };
+}
+
+interface CompositeAggregationsResponse {
+  groupings: {
+    buckets: Aggregation[];
+  };
+}
+
+const getCurrentValueFromAggregations = (
+  aggregations: Aggregation,
+  aggType: MetricExpressionParams['aggType']
+) => {
+  try {
+    const { buckets } = aggregations.aggregatedIntervals;
+    if (!buckets.length) return null; // No Data state
+    const mostRecentBucket = buckets[buckets.length - 1];
+    if (aggType === 'count') {
+      return mostRecentBucket.doc_count;
+    }
+    const { value } = mostRecentBucket.aggregatedValue;
+    return value;
+  } catch (e) {
+    return undefined; // Error state
+  }
+};
+
+const getParsedFilterQuery: (
+  filterQuery: string | undefined
+) => Record<string, any> = filterQuery => {
+  if (!filterQuery) return {};
+  try {
+    return JSON.parse(filterQuery).bool;
+  } catch (e) {
+    return {
+      query_string: {
+        query: filterQuery,
+        analyze_wildcard: true,
+      },
+    };
+  }
+};
+
+export const getElasticsearchMetricQuery = (
+  { metric, aggType, timeUnit, timeSize }: MetricExpressionParams,
+  groupBy?: string,
+  filterQuery?: string
+) => {
+  const interval = `${timeSize}${timeUnit}`;
+
+  const aggregations =
+    aggType === 'count'
+      ? {}
+      : aggType === 'rate'
+      ? networkTraffic('aggregatedValue', metric)
+      : {
+          aggregatedValue: {
+            [aggType]: {
+              field: metric,
+            },
+          },
+        };
+
+  const baseAggs = {
+    aggregatedIntervals: {
+      date_histogram: {
+        field: '@timestamp',
+        fixed_interval: interval,
+      },
+      aggregations,
+    },
+  };
+
+  const aggs = groupBy
+    ? {
+        groupings: {
+          composite: {
+            size: 10,
+            sources: [
+              {
+                groupBy: {
+                  terms: {
+                    field: groupBy,
+                  },
+                },
+              },
+            ],
+          },
+          aggs: baseAggs,
+        },
+      }
+    : baseAggs;
+
+  const parsedFilterQuery = getParsedFilterQuery(filterQuery);
+
+  return {
+    query: {
+      bool: {
+        filter: [
+          {
+            range: {
+              '@timestamp': {
+                gte: `now-${interval}`,
+              },
+            },
+          },
+          {
+            exists: {
+              field: metric,
+            },
+          },
+        ],
+        ...parsedFilterQuery,
+      },
+    },
+    size: 0,
+    aggs,
+  };
+};
+
+const getMetric: (
+  services: AlertServices,
+  params: MetricExpressionParams,
+  groupBy: string | undefined,
+  filterQuery: string | undefined
+) => Promise<Record<string, number>> = async function(
+  { callCluster },
+  params,
+  groupBy,
+  filterQuery
+) {
+  const { indexPattern, aggType } = params;
+  const searchBody = getElasticsearchMetricQuery(params, groupBy, filterQuery);
+
+  try {
+    if (groupBy) {
+      const bucketSelector = (
+        response: InfraDatabaseSearchResponse<{}, CompositeAggregationsResponse>
+      ) => response.aggregations?.groupings?.buckets || [];
+      const afterKeyHandler = createAfterKeyHandler(
+        'aggs.groupings.composite.after',
+        response => response.aggregations?.groupings?.after_key
+      );
+      const compositeBuckets = (await getAllCompositeData(
+        body => callCluster('search', { body, index: indexPattern }),
+        searchBody,
+        bucketSelector,
+        afterKeyHandler
+      )) as Array<Aggregation & { key: { groupBy: string } }>;
+      return compositeBuckets.reduce(
+        (result, bucket) => ({
+          ...result,
+          [bucket.key.groupBy]: getCurrentValueFromAggregations(bucket, aggType),
+        }),
+        {}
+      );
+    }
+    const result = await callCluster('search', {
+      body: searchBody,
+      index: indexPattern,
+    });
+    return { '*': getCurrentValueFromAggregations(result.aggregations, aggType) };
+  } catch (e) {
+    return { '*': undefined }; // Trigger an Error state
+  }
+};
+
+const comparatorMap = {
+  [Comparator.BETWEEN]: (value: number, [a, b]: number[]) =>
+    value >= Math.min(a, b) && value <= Math.max(a, b),
+  // `threshold` is always an array of numbers in case the BETWEEN comparator is
+  // used; all other compartors will just destructure the first value in the array
+  [Comparator.GT]: (a: number, [b]: number[]) => a > b,
+  [Comparator.LT]: (a: number, [b]: number[]) => a < b,
+  [Comparator.GT_OR_EQ]: (a: number, [b]: number[]) => a >= b,
+  [Comparator.LT_OR_EQ]: (a: number, [b]: number[]) => a <= b,
+};
+
+export const createMetricThresholdExecutor = (alertUUID: string) =>
+  async function({ services, params }: AlertExecutorOptions) {
+    const { criteria, groupBy, filterQuery } = params as {
+      criteria: MetricExpressionParams[];
+      groupBy: string | undefined;
+      filterQuery: string | undefined;
+    };
+
+    const alertResults = await Promise.all(
+      criteria.map(criterion =>
+        (async () => {
+          const currentValues = await getMetric(services, criterion, groupBy, filterQuery);
+          const { threshold, comparator } = criterion;
+          const comparisonFunction = comparatorMap[comparator];
+          return mapValues(currentValues, value => ({
+            shouldFire:
+              value !== undefined && value !== null && comparisonFunction(value, threshold),
+            currentValue: value,
+            isNoData: value === null,
+            isError: value === undefined,
+          }));
+        })()
+      )
+    );
+
+    const groups = Object.keys(alertResults[0]);
+    for (const group of groups) {
+      const alertInstance = services.alertInstanceFactory(`${alertUUID}-${group}`);
+
+      // AND logic; all criteria must be across the threshold
+      const shouldAlertFire = alertResults.every(result => result[group].shouldFire);
+      // AND logic; because we need to evaluate all criteria, if one of them reports no data then the
+      // whole alert is in a No Data/Error state
+      const isNoData = alertResults.some(result => result[group].isNoData);
+      const isError = alertResults.some(result => result[group].isError);
+      if (shouldAlertFire) {
+        alertInstance.scheduleActions(FIRED_ACTIONS.id, {
+          group,
+          value: alertResults.map(result => result[group].currentValue),
+        });
+      }
+      // Future use: ability to fetch display current alert state
+      alertInstance.replaceState({
+        alertState: isError
+          ? AlertStates.ERROR
+          : isNoData
+          ? AlertStates.NO_DATA
+          : shouldAlertFire
+          ? AlertStates.ALERT
+          : AlertStates.OK,
+      });
+    }
+  };
+
+export const FIRED_ACTIONS = {
+  id: 'metrics.threshold.fired',
+  name: i18n.translate('xpack.infra.metrics.alerting.threshold.fired', {
+    defaultMessage: 'Fired',
+  }),
+};

x-pack/plugins/infra/server/lib/alerting/metric_threshold/register_metric_threshold_alert_type.ts

@@ -4,188 +4,10 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 import uuid from 'uuid';
-import { mapValues } from 'lodash';
-import { i18n } from '@kbn/i18n';
 import { schema } from '@kbn/config-schema';
-import { InfraDatabaseSearchResponse } from '../../adapters/framework/adapter_types';
-import { createAfterKeyHandler } from '../../../utils/create_afterkey_handler';
-import { getAllCompositeData } from '../../../utils/get_all_composite_data';
-import { networkTraffic } from '../../../../common/inventory_models/shared/metrics/snapshot/network_traffic';
-import {
-  MetricExpressionParams,
-  Comparator,
-  AlertStates,
-  METRIC_THRESHOLD_ALERT_TYPE_ID,
-} from './types';
-import { AlertServices, PluginSetupContract } from '../../../../../alerting/server';
-
-interface Aggregation {
-  aggregatedIntervals: { buckets: Array<{ aggregatedValue: { value: number } }> };
-}
-
-interface CompositeAggregationsResponse {
-  groupings: {
-    buckets: Aggregation[];
-  };
-}
-
-const FIRED_ACTIONS = {
-  id: 'metrics.threshold.fired',
-  name: i18n.translate('xpack.infra.metrics.alerting.threshold.fired', {
-    defaultMessage: 'Fired',
-  }),
-};
-
-const getCurrentValueFromAggregations = (aggregations: Aggregation) => {
-  try {
-    const { buckets } = aggregations.aggregatedIntervals;
-    if (!buckets.length) return null; // No Data state
-    const { value } = buckets[buckets.length - 1].aggregatedValue;
-    return value;
-  } catch (e) {
-    return undefined; // Error state
-  }
-};
-
-const getParsedFilterQuery: (
-  filterQuery: string | undefined
-) => Record<string, any> = filterQuery => {
-  if (!filterQuery) return {};
-  try {
-    return JSON.parse(filterQuery).bool;
-  } catch (e) {
-    return {
-      query_string: {
-        query: filterQuery,
-        analyze_wildcard: true,
-      },
-    };
-  }
-};
-
-const getMetric: (
-  services: AlertServices,
-  params: MetricExpressionParams,
-  groupBy: string | undefined,
-  filterQuery: string | undefined
-) => Promise<Record<string, number>> = async function(
-  { callCluster },
-  { metric, aggType, timeUnit, timeSize, indexPattern },
-  groupBy,
-  filterQuery
-) {
-  const interval = `${timeSize}${timeUnit}`;
-
-  const aggregations =
-    aggType === 'rate'
-      ? networkTraffic('aggregatedValue', metric)
-      : {
-          aggregatedValue: {
-            [aggType]: {
-              field: metric,
-            },
-          },
-        };
-
-  const baseAggs = {
-    aggregatedIntervals: {
-      date_histogram: {
-        field: '@timestamp',
-        fixed_interval: interval,
-      },
-      aggregations,
-    },
-  };
-
-  const aggs = groupBy
-    ? {
-        groupings: {
-          composite: {
-            size: 10,
-            sources: [
-              {
-                groupBy: {
-                  terms: {
-                    field: groupBy,
-                  },
-                },
-              },
-            ],
-          },
-          aggs: baseAggs,
-        },
-      }
-    : baseAggs;
-
-  const parsedFilterQuery = getParsedFilterQuery(filterQuery);
-
-  const searchBody = {
-    query: {
-      bool: {
-        filter: [
-          {
-            range: {
-              '@timestamp': {
-                gte: `now-${interval}`,
-              },
-            },
-          },
-          {
-            exists: {
-              field: metric,
-            },
-          },
-        ],
-        ...parsedFilterQuery,
-      },
-    },
-    size: 0,
-    aggs,
-  };
-
-  try {
-    if (groupBy) {
-      const bucketSelector = (
-        response: InfraDatabaseSearchResponse<{}, CompositeAggregationsResponse>
-      ) => response.aggregations?.groupings?.buckets || [];
-      const afterKeyHandler = createAfterKeyHandler(
-        'aggs.groupings.composite.after',
-        response => response.aggregations?.groupings?.after_key
-      );
-      const compositeBuckets = (await getAllCompositeData(
-        body => callCluster('search', { body, index: indexPattern }),
-        searchBody,
-        bucketSelector,
-        afterKeyHandler
-      )) as Array<Aggregation & { key: { groupBy: string } }>;
-      return compositeBuckets.reduce(
-        (result, bucket) => ({
-          ...result,
-          [bucket.key.groupBy]: getCurrentValueFromAggregations(bucket),
-        }),
-        {}
-      );
-    }
-    const result = await callCluster('search', {
-      body: searchBody,
-      index: indexPattern,
-    });
-    return { '*': getCurrentValueFromAggregations(result.aggregations) };
-  } catch (e) {
-    return { '*': undefined }; // Trigger an Error state
-  }
-};
-
-const comparatorMap = {
-  [Comparator.BETWEEN]: (value: number, [a, b]: number[]) =>
-    value >= Math.min(a, b) && value <= Math.max(a, b),
-  // `threshold` is always an array of numbers in case the BETWEEN comparator is
-  // used; all other compartors will just destructure the first value in the array
-  [Comparator.GT]: (a: number, [b]: number[]) => a > b,
-  [Comparator.LT]: (a: number, [b]: number[]) => a < b,
-  [Comparator.GT_OR_EQ]: (a: number, [b]: number[]) => a >= b,
-  [Comparator.LT_OR_EQ]: (a: number, [b]: number[]) => a <= b,
-};
+import { PluginSetupContract } from '../../../../../alerting/server';
+import { createMetricThresholdExecutor, FIRED_ACTIONS } from './metric_threshold_executor';
+import { METRIC_THRESHOLD_ALERT_TYPE_ID } from './types';
 
 export async function registerMetricThresholdAlertType(alertingPlugin: PluginSetupContract) {
   if (!alertingPlugin) {
@@ -217,59 +39,6 @@ export async function registerMetricThresholdAlertType(alertingPlugin: PluginSet
     },
     defaultActionGroupId: FIRED_ACTIONS.id,
     actionGroups: [FIRED_ACTIONS],
-    async executor({ services, params }) {
-      const { criteria, groupBy, filterQuery } = params as {
-        criteria: MetricExpressionParams[];
-        groupBy: string | undefined;
-        filterQuery: string | undefined;
-      };
-
-      const alertResults = await Promise.all(
-        criteria.map(criterion =>
-          (async () => {
-            const currentValues = await getMetric(services, criterion, groupBy, filterQuery);
-            const { threshold, comparator } = criterion;
-            const comparisonFunction = comparatorMap[comparator];
-
-            return mapValues(currentValues, value => ({
-              shouldFire:
-                value !== undefined && value !== null && comparisonFunction(value, threshold),
-              currentValue: value,
-              isNoData: value === null,
-              isError: value === undefined,
-            }));
-          })()
-        )
-      );
-
-      const groups = Object.keys(alertResults[0]);
-      for (const group of groups) {
-        const alertInstance = services.alertInstanceFactory(`${alertUUID}-${group}`);
-
-        // AND logic; all criteria must be across the threshold
-        const shouldAlertFire = alertResults.every(result => result[group].shouldFire);
-        // AND logic; because we need to evaluate all criteria, if one of them reports no data then the
-        // whole alert is in a No Data/Error state
-        const isNoData = alertResults.some(result => result[group].isNoData);
-        const isError = alertResults.some(result => result[group].isError);
-        if (shouldAlertFire) {
-          alertInstance.scheduleActions(FIRED_ACTIONS.id, {
-            group,
-            value: alertResults.map(result => result[group].currentValue),
-          });
-        }
-
-        // Future use: ability to fetch display current alert state
-        alertInstance.replaceState({
-          alertState: isError
-            ? AlertStates.ERROR
-            : isNoData
-            ? AlertStates.NO_DATA
-            : shouldAlertFire
-            ? AlertStates.ALERT
-            : AlertStates.OK,
-        });
-      }
-    },
+    executor: createMetricThresholdExecutor(alertUUID),
   });
 }

x-pack/plugins/infra/server/lib/alerting/metric_threshold/test_mocks.ts

@@ -0,0 +1,110 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+const bucketsA = [
+  {
+    doc_count: 2,
+    aggregatedValue: { value: 0.5 },
+  },
+  {
+    doc_count: 3,
+    aggregatedValue: { value: 1.0 },
+  },
+];
+
+const bucketsB = [
+  {
+    doc_count: 4,
+    aggregatedValue: { value: 2.5 },
+  },
+  {
+    doc_count: 5,
+    aggregatedValue: { value: 3.5 },
+  },
+];
+
+export const basicMetricResponse = {
+  aggregations: {
+    aggregatedIntervals: {
+      buckets: bucketsA,
+    },
+  },
+};
+
+export const alternateMetricResponse = {
+  aggregations: {
+    aggregatedIntervals: {
+      buckets: bucketsB,
+    },
+  },
+};
+
+export const basicCompositeResponse = {
+  aggregations: {
+    groupings: {
+      after_key: 'foo',
+      buckets: [
+        {
+          key: {
+            groupBy: 'a',
+          },
+          aggregatedIntervals: {
+            buckets: bucketsA,
+          },
+        },
+        {
+          key: {
+            groupBy: 'b',
+          },
+          aggregatedIntervals: {
+            buckets: bucketsB,
+          },
+        },
+      ],
+    },
+  },
+  hits: {
+    total: {
+      value: 2,
+    },
+  },
+};
+
+export const alternateCompositeResponse = {
+  aggregations: {
+    groupings: {
+      after_key: 'foo',
+      buckets: [
+        {
+          key: {
+            groupBy: 'a',
+          },
+          aggregatedIntervals: {
+            buckets: bucketsB,
+          },
+        },
+        {
+          key: {
+            groupBy: 'b',
+          },
+          aggregatedIntervals: {
+            buckets: bucketsA,
+          },
+        },
+      ],
+    },
+  },
+  hits: {
+    total: {
+      value: 2,
+    },
+  },
+};
+
+export const compositeEndResponse = {
+  aggregations: {},
+  hits: { total: { value: 0 } },
+};

x-pack/plugins/infra/server/lib/alerting/metric_threshold/types.ts

@@ -33,5 +33,4 @@ export interface MetricExpressionParams {
   indexPattern: string;
   threshold: number[];
   comparator: Comparator;
-  filterQuery: string;
 }

x-pack/test/api_integration/apis/infra/index.js

@@ -16,6 +16,7 @@ export default function({ loadTestFile }) {
     loadTestFile(require.resolve('./sources'));
     loadTestFile(require.resolve('./waffle'));
     loadTestFile(require.resolve('./log_item'));
+    loadTestFile(require.resolve('./metrics_alerting'));
     loadTestFile(require.resolve('./metrics_explorer'));
     loadTestFile(require.resolve('./feature_controls'));
     loadTestFile(require.resolve('./ip_to_hostname'));

x-pack/test/api_integration/apis/infra/metrics_alerting.ts

@@ -0,0 +1,98 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import expect from '@kbn/expect';
+import { getElasticsearchMetricQuery } from '../../../../plugins/infra/server/lib/alerting/metric_threshold/metric_threshold_executor';
+import { MetricExpressionParams } from '../../../../plugins/infra/server/lib/alerting/metric_threshold/types';
+
+import { FtrProviderContext } from '../../ftr_provider_context';
+
+export default function({ getService }: FtrProviderContext) {
+  const client = getService('legacyEs');
+  const index = 'test-index';
+  const baseParams = {
+    metric: 'test.metric',
+    timeUnit: 'm',
+    timeSize: 5,
+  };
+  describe('Metrics Threshold Alerts', () => {
+    before(async () => {
+      await client.index({
+        index,
+        body: {},
+      });
+    });
+    const aggs = ['avg', 'min', 'max', 'rate', 'cardinality', 'count'];
+
+    describe('querying the entire infrastructure', () => {
+      for (const aggType of aggs) {
+        it(`should work with the ${aggType} aggregator`, async () => {
+          const searchBody = getElasticsearchMetricQuery({
+            ...baseParams,
+            aggType,
+          } as MetricExpressionParams);
+          const result = await client.search({
+            index,
+            body: searchBody,
+          });
+          expect(result.error).to.not.be.ok();
+          expect(result.hits).to.be.ok();
+        });
+      }
+      it('should work with a filterQuery', async () => {
+        const searchBody = getElasticsearchMetricQuery(
+          {
+            ...baseParams,
+            aggType: 'avg',
+          } as MetricExpressionParams,
+          undefined,
+          '{"bool":{"should":[{"match_phrase":{"agent.hostname":"foo"}}],"minimum_should_match":1}}'
+        );
+        const result = await client.search({
+          index,
+          body: searchBody,
+        });
+        expect(result.error).to.not.be.ok();
+        expect(result.hits).to.be.ok();
+      });
+    });
+    describe('querying with a groupBy parameter', () => {
+      for (const aggType of aggs) {
+        it(`should work with the ${aggType} aggregator`, async () => {
+          const searchBody = getElasticsearchMetricQuery(
+            {
+              ...baseParams,
+              aggType,
+            } as MetricExpressionParams,
+            'agent.id'
+          );
+          const result = await client.search({
+            index,
+            body: searchBody,
+          });
+          expect(result.error).to.not.be.ok();
+          expect(result.hits).to.be.ok();
+        });
+      }
+      it('should work with a filterQuery', async () => {
+        const searchBody = getElasticsearchMetricQuery(
+          {
+            ...baseParams,
+            aggType: 'avg',
+          } as MetricExpressionParams,
+          'agent.id',
+          '{"bool":{"should":[{"match_phrase":{"agent.hostname":"foo"}}],"minimum_should_match":1}}'
+        );
+        const result = await client.search({
+          index,
+          body: searchBody,
+        });
+        expect(result.error).to.not.be.ok();
+        expect(result.hits).to.be.ok();
+      });
+    });
+  });
+}