[Metrics Alerts] Create Metric Threshold Alert Type and Executor

x-pack/plugins/infra/kibana.json

@@ -2,7 +2,17 @@
   "id": "infra",
   "version": "8.0.0",
   "kibanaVersion": "kibana",
-  "requiredPlugins": ["features", "apm", "usageCollection", "spaces", "home", "data", "data_enhanced", "metrics"],
+  "requiredPlugins": [
+    "features",
+    "apm",
+    "usageCollection",
+    "spaces",
+    "home",
+    "data",
+    "data_enhanced",
+    "metrics",
+    "alerting"
+  ],
   "server": true,
   "ui": true,
   "configPath": ["xpack", "infra"]

x-pack/plugins/infra/server/lib/adapters/framework/adapter_types.ts

@@ -13,6 +13,7 @@ import { SpacesPluginSetup } from '../../../../../../plugins/spaces/server';
 import { VisTypeTimeseriesSetup } from '../../../../../../../src/plugins/vis_type_timeseries/server';
 import { APMPluginContract } from '../../../../../../plugins/apm/server';
 import { HomeServerPluginSetup } from '../../../../../../../src/plugins/home/server';
+import { PluginSetupContract as AlertingPluginContract } from '../../../../../../plugins/alerting/server';
 
 // NP_TODO: Compose real types from plugins we depend on, no "any"
 export interface InfraServerPluginDeps {
@@ -25,6 +26,7 @@ export interface InfraServerPluginDeps {
   };
   features: FeaturesPluginSetup;
   apm: APMPluginContract;
+  alerting: AlertingPluginContract;
 }
 
 export interface CallWithRequestParams extends GenericParams {

x-pack/plugins/infra/server/lib/alerting/index.ts

@@ -0,0 +1,7 @@
+/*
+ * 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.
+ */
+
+export { registerAlertTypes } from './register_alert_types';

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

@@ -0,0 +1,132 @@
+/*
+ * 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 uuid from 'uuid';
+import { i18n } from '@kbn/i18n';
+import { schema } from '@kbn/config-schema';
+import {
+  MetricThresholdAlertTypeParams,
+  Comparator,
+  AlertStates,
+  METRIC_THRESHOLD_ALERT_TYPE_ID,
+} from './types';
+import { AlertServices, PluginSetupContract } from '../../../../../alerting/server';
+
+const FIRED_ACTIONS = {
+  id: 'metrics.threshold.fired',
+  name: i18n.translate('xpack.infra.metrics.alerting.threshold.fired', {
+    defaultMessage: 'Fired',
+  }),
+};
+
+async function getMetric(
+  { callCluster }: AlertServices,
+  { metric, aggType, timeUnit, timeSize, indexPattern }: MetricThresholdAlertTypeParams
+) {
+  const interval = `${timeSize}${timeUnit}`;
+  const searchBody = {
+    query: {
+      bool: {
+        filter: [
+          {
+            range: {
+              '@timestamp': {
+                gte: `now-${interval}`,
+              },
+            },
+            exists: {
+              field: metric,
+            },
+          },
+        ],
+      },
+    },
+    size: 0,
+    aggs: {
+      aggregatedIntervals: {
+        date_histogram: {
+          field: '@timestamp',
+          fixed_interval: interval,
+        },
+        aggregations: {
+          aggregatedValue: {
+            [aggType]: {
+              field: metric,
+            },
+          },
+        },
+      },
+    },
+  };
+
+  const result = await callCluster('search', {
+    body: searchBody,
+    index: indexPattern,
+  });
+
+  const { buckets } = result.aggregations.aggregatedIntervals;
+  const { value } = buckets[buckets.length - 1].aggregatedValue;
+  return value;
+}
+
+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 async function registerMetricThresholdAlertType(alertingPlugin: PluginSetupContract) {
+  if (!alertingPlugin) {
+    throw new Error(
+      'Cannot register metric threshold alert type.  Both the actions and alerting plugins need to be enabled.'
+    );
+  }
+  const alertUUID = uuid.v4();
+
+  alertingPlugin.registerType({
+    id: METRIC_THRESHOLD_ALERT_TYPE_ID,
+    name: 'Metric Alert - Threshold',
+    validate: {
+      params: schema.object({
+        threshold: schema.arrayOf(schema.number()),
+        comparator: schema.string(),
+        aggType: schema.string(),
+        metric: schema.string(),
+        timeUnit: schema.string(),
+        timeSize: schema.number(),
+        indexPattern: schema.string(),
+      }),
+    },
+    defaultActionGroupId: FIRED_ACTIONS.id,
+    actionGroups: [FIRED_ACTIONS],
+    async executor({ services, params }) {
+      const { threshold, comparator } = params as MetricThresholdAlertTypeParams;
+      const alertInstance = services.alertInstanceFactory(alertUUID);
+      const currentValue = await getMetric(services, params as MetricThresholdAlertTypeParams);
+      if (typeof currentValue === 'undefined')
+        throw new Error('Could not get current value of metric');
+
+      const comparisonFunction = comparatorMap[comparator];
+
+      const isValueInAlertState = comparisonFunction(currentValue, threshold);
+
+      if (isValueInAlertState) {
+        alertInstance.scheduleActions(FIRED_ACTIONS.id, {
+          value: currentValue,
+        });
+      }
+
+      // Future use: ability to fetch display current alert state
+      alertInstance.replaceState({
+        alertState: isValueInAlertState ? AlertStates.ALERT : AlertStates.OK,
+      });
+    },
+  });
+}

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

@@ -0,0 +1,34 @@
+/*
+ * 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 { MetricsExplorerAggregation } from '../../../../common/http_api/metrics_explorer';
+
+export const METRIC_THRESHOLD_ALERT_TYPE_ID = 'metrics.alert.threshold';
+
+export enum Comparator {
+  GT = '>',
+  LT = '<',
+  GT_OR_EQ = '>=',
+  LT_OR_EQ = '<=',
+  BETWEEN = 'between',
+}
+
+export enum AlertStates {
+  OK,
+  ALERT,
+}
+
+export type TimeUnit = 's' | 'm' | 'h' | 'd';
+
+export interface MetricThresholdAlertTypeParams {
+  aggType: MetricsExplorerAggregation;
+  metric: string;
+  timeSize: number;
+  timeUnit: TimeUnit;
+  indexPattern: string;
+  threshold: number[];
+  comparator: Comparator;
+}

x-pack/plugins/infra/server/lib/alerting/register_alert_types.ts

@@ -0,0 +1,20 @@
+/*
+ * 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 { PluginSetupContract } from '../../../../alerting/server';
+import { registerMetricThresholdAlertType } from './metric_threshold/register_metric_threshold_alert_type';
+
+const registerAlertTypes = (alertingPlugin: PluginSetupContract) => {
+  if (alertingPlugin) {
+    const registerFns = [registerMetricThresholdAlertType];
+
+    registerFns.forEach(fn => {
+      fn(alertingPlugin);
+    });
+  }
+};
+
+export { registerAlertTypes };

x-pack/plugins/infra/server/plugin.ts

@@ -27,6 +27,7 @@ import { InfraServerPluginDeps } from './lib/adapters/framework';
 import { METRICS_FEATURE, LOGS_FEATURE } from './features';
 import { UsageCollector } from './usage/usage_collector';
 import { InfraStaticSourceConfiguration } from './lib/sources/types';
+import { registerAlertTypes } from './lib/alerting';
 
 export const config = {
   schema: schema.object({
@@ -146,6 +147,7 @@ export class InfraServerPlugin {
     ]);
 
     initInfraServer(this.libs);
+    registerAlertTypes(plugins.alerting);
 
     // Telemetry
     UsageCollector.registerUsageCollector(plugins.usageCollection);