[Response Ops][Alerting] Update stack rules to respect max alert limit (#141000)

* wip

* wip

* Adding bucket selector clauses

* Adding comparator script generator

* Generating all the right queries

* Skip condition check if group agg

* Fixing functional test

* Fixing comparator script

* Fixing tests

* Fixing tests

* Renaming

* Using limit services in es query rule executor

Co-authored-by: Kibana Machine <42973632+kibanamachine@users.noreply.github.com>

x-pack/plugins/stack_alerts/server/alert_types/es_query/executor.ts

@@ -29,6 +29,8 @@ export async function executor(
   const currentTimestamp = new Date().toISOString();
   const publicBaseUrl = core.http.basePath.publicBaseUrl ?? '';
 
+  const alertLimit = alertFactory.alertLimit.getValue();
+
   const compareFn = ComparatorFns.get(params.thresholdComparator);
   if (compareFn == null) {
     throw new Error(getInvalidComparatorError(params.thresholdComparator));
@@ -91,6 +93,12 @@ export async function executor(
     if (firstValidTimefieldSort) {
       latestTimestamp = firstValidTimefieldSort;
     }
+
+    // we only create one alert if the condition is met, so we would only ever
+    // reach the alert limit if the limit is less than 1
+    alertFactory.alertLimit.setLimitReached(alertLimit < 1);
+  } else {
+    alertFactory.alertLimit.setLimitReached(false);
   }
 
   const { getRecoveredAlerts } = alertFactory.done();

x-pack/plugins/stack_alerts/server/alert_types/index_threshold/action_context.test.ts

@@ -6,7 +6,7 @@
  */
 
 import { BaseActionContext, addMessages } from './action_context';
-import { ParamsSchema } from './alert_type_params';
+import { ParamsSchema } from './rule_type_params';
 
 describe('ActionContext', () => {
   it('generates expected properties if aggField is null', async () => {
@@ -28,10 +28,10 @@ describe('ActionContext', () => {
       value: 42,
       conditions: 'count greater than 4',
     };
-    const context = addMessages({ name: '[alert-name]' }, base, params);
-    expect(context.title).toMatchInlineSnapshot(`"alert [alert-name] group [group] met threshold"`);
+    const context = addMessages({ name: '[rule-name]' }, base, params);
+    expect(context.title).toMatchInlineSnapshot(`"alert [rule-name] group [group] met threshold"`);
     expect(context.message).toEqual(
-      `alert '[alert-name]' is active for group '[group]':
+      `alert '[rule-name]' is active for group '[group]':
 
 - Value: 42
 - Conditions Met: count greater than 4 over 5m
@@ -59,10 +59,10 @@ describe('ActionContext', () => {
       value: 42,
       conditions: 'avg([aggField]) greater than 4.2',
     };
-    const context = addMessages({ name: '[alert-name]' }, base, params);
-    expect(context.title).toMatchInlineSnapshot(`"alert [alert-name] group [group] met threshold"`);
+    const context = addMessages({ name: '[rule-name]' }, base, params);
+    expect(context.title).toMatchInlineSnapshot(`"alert [rule-name] group [group] met threshold"`);
     expect(context.message).toEqual(
-      `alert '[alert-name]' is active for group '[group]':
+      `alert '[rule-name]' is active for group '[group]':
 
 - Value: 42
 - Conditions Met: avg([aggField]) greater than 4.2 over 5m
@@ -89,10 +89,10 @@ describe('ActionContext', () => {
       value: 4,
       conditions: 'count between 4 and 5',
     };
-    const context = addMessages({ name: '[alert-name]' }, base, params);
-    expect(context.title).toMatchInlineSnapshot(`"alert [alert-name] group [group] met threshold"`);
+    const context = addMessages({ name: '[rule-name]' }, base, params);
+    expect(context.title).toMatchInlineSnapshot(`"alert [rule-name] group [group] met threshold"`);
     expect(context.message).toEqual(
-      `alert '[alert-name]' is active for group '[group]':
+      `alert '[rule-name]' is active for group '[group]':
 
 - Value: 4
 - Conditions Met: count between 4 and 5 over 5m
@@ -119,10 +119,10 @@ describe('ActionContext', () => {
       value: 'unknown',
       conditions: 'count between 4 and 5',
     };
-    const context = addMessages({ name: '[alert-name]' }, base, params);
-    expect(context.title).toMatchInlineSnapshot(`"alert [alert-name] group [group] met threshold"`);
+    const context = addMessages({ name: '[rule-name]' }, base, params);
+    expect(context.title).toMatchInlineSnapshot(`"alert [rule-name] group [group] met threshold"`);
     expect(context.message).toEqual(
-      `alert '[alert-name]' is active for group '[group]':
+      `alert '[rule-name]' is active for group '[group]':
 
 - Value: unknown
 - Conditions Met: count between 4 and 5 over 5m

x-pack/plugins/stack_alerts/server/alert_types/index_threshold/action_context.ts

@@ -7,9 +7,9 @@
 
 import { i18n } from '@kbn/i18n';
 import { RuleExecutorOptions, AlertInstanceContext } from '@kbn/alerting-plugin/server';
-import { Params } from './alert_type_params';
+import { Params } from './rule_type_params';
 
-// alert type context provided to actions
+// rule type context provided to actions
 
 type RuleInfo = Pick<RuleExecutorOptions, 'name'>;
 
@@ -21,10 +21,10 @@ export interface ActionContext extends BaseActionContext {
 }
 
 export interface BaseActionContext extends AlertInstanceContext {
-  // the aggType used in the alert
+  // the aggType used in the rule
   // the value of the aggField, if used, otherwise 'all documents'
   group: string;
-  // the date the alert was run as an ISO date
+  // the date the rule was run as an ISO date
   date: string;
   // the value that met the threshold
   value: number | string;

x-pack/plugins/stack_alerts/server/alert_types/index_threshold/index.ts

@@ -7,7 +7,7 @@
 
 import { Logger } from '@kbn/core/server';
 import { AlertingSetup, StackAlertsStartDeps } from '../../types';
-import { getAlertType } from './alert_type';
+import { getRuleType } from './rule_type';
 
 // future enhancement: make these configurable?
 export const MAX_INTERVALS = 1000;
@@ -22,5 +22,5 @@ interface RegisterParams {
 
 export function register(params: RegisterParams) {
   const { logger, data, alerting } = params;
-  alerting.registerType(getAlertType(logger, data));
+  alerting.registerType(getRuleType(logger, data));
 }

x-pack/plugins/stack_alerts/server/alert_types/index_threshold/rule_type.test.ts

@@ -6,34 +6,44 @@
  */
 
 import uuid from 'uuid';
+import sinon from 'sinon';
 import type { Writable } from '@kbn/utility-types';
 import { loggingSystemMock } from '@kbn/core/server/mocks';
 import { RuleExecutorServices } from '@kbn/alerting-plugin/server';
-import { getAlertType, ActionGroupId } from './alert_type';
+import { getRuleType, ActionGroupId } from './rule_type';
 import { ActionContext } from './action_context';
-import { Params } from './alert_type_params';
+import { Params } from './rule_type_params';
+import { TIME_SERIES_BUCKET_SELECTOR_FIELD } from '@kbn/triggers-actions-ui-plugin/server';
 import { RuleExecutorServicesMock, alertsMock } from '@kbn/alerting-plugin/server/mocks';
 import { Comparator } from '../../../common/comparator_types';
 
-describe('alertType', () => {
+let fakeTimer: sinon.SinonFakeTimers;
+
+describe('ruleType', () => {
   const logger = loggingSystemMock.create().get();
   const data = {
     timeSeriesQuery: jest.fn(),
   };
   const alertServices: RuleExecutorServicesMock = alertsMock.createRuleExecutorServices();
 
-  const alertType = getAlertType(logger, Promise.resolve(data));
+  const ruleType = getRuleType(logger, Promise.resolve(data));
+
+  beforeAll(() => {
+    fakeTimer = sinon.useFakeTimers();
+  });
 
   afterEach(() => {
     data.timeSeriesQuery.mockReset();
   });
 
-  it('alert type creation structure is the expected value', async () => {
-    expect(alertType.id).toBe('.index-threshold');
-    expect(alertType.name).toBe('Index threshold');
-    expect(alertType.actionGroups).toEqual([{ id: 'threshold met', name: 'Threshold met' }]);
+  afterAll(() => fakeTimer.restore());
 
-    expect(alertType.actionVariables).toMatchInlineSnapshot(`
+  it('rule type creation structure is the expected value', async () => {
+    expect(ruleType.id).toBe('.index-threshold');
+    expect(ruleType.name).toBe('Index threshold');
+    expect(ruleType.actionGroups).toEqual([{ id: 'threshold met', name: 'Threshold met' }]);
+
+    expect(ruleType.actionVariables).toMatchInlineSnapshot(`
       Object {
         "context": Array [
           Object {
@@ -123,11 +133,11 @@ describe('alertType', () => {
       threshold: [0],
     };
 
-    expect(alertType.validate?.params?.validate(params)).toBeTruthy();
+    expect(ruleType.validate?.params?.validate(params)).toBeTruthy();
   });
 
   it('validator fails with invalid params', async () => {
-    const paramsSchema = alertType.validate?.params;
+    const paramsSchema = ruleType.validate?.params;
     if (!paramsSchema) throw new Error('params validator not set');
 
     const params: Partial<Writable<Params>> = {
@@ -168,7 +178,7 @@ describe('alertType', () => {
       threshold: [1],
     };
 
-    await alertType.executor({
+    await ruleType.executor({
       alertId: uuid.v4(),
       executionId: uuid.v4(),
       startedAt: new Date(),
@@ -234,7 +244,7 @@ describe('alertType', () => {
       threshold: [1],
     };
 
-    await alertType.executor({
+    await ruleType.executor({
       alertId: uuid.v4(),
       executionId: uuid.v4(),
       startedAt: new Date(),
@@ -300,7 +310,7 @@ describe('alertType', () => {
       threshold: [1],
     };
 
-    await alertType.executor({
+    await ruleType.executor({
       alertId: uuid.v4(),
       executionId: uuid.v4(),
       startedAt: new Date(),
@@ -342,4 +352,90 @@ describe('alertType', () => {
 
     expect(customAlertServices.alertFactory.create).not.toHaveBeenCalled();
   });
+
+  it('should correctly pass comparator script to timeSeriesQuery', async () => {
+    data.timeSeriesQuery.mockImplementation((...args) => {
+      return {
+        results: [
+          {
+            group: 'all documents',
+            metrics: [['2021-07-14T14:49:30.978Z', 0]],
+          },
+        ],
+      };
+    });
+    const params: Params = {
+      index: 'index-name',
+      timeField: 'time-field',
+      aggType: 'foo',
+      groupBy: 'all',
+      timeWindowSize: 5,
+      timeWindowUnit: 'm',
+      thresholdComparator: Comparator.LT,
+      threshold: [1],
+    };
+
+    await ruleType.executor({
+      alertId: uuid.v4(),
+      executionId: uuid.v4(),
+      startedAt: new Date(),
+      previousStartedAt: new Date(),
+      services: alertServices as unknown as RuleExecutorServices<
+        {},
+        ActionContext,
+        typeof ActionGroupId
+      >,
+      params,
+      state: {
+        latestTimestamp: undefined,
+      },
+      spaceId: uuid.v4(),
+      name: uuid.v4(),
+      tags: [],
+      createdBy: null,
+      updatedBy: null,
+      rule: {
+        name: uuid.v4(),
+        tags: [],
+        consumer: '',
+        producer: '',
+        ruleTypeId: '',
+        ruleTypeName: '',
+        enabled: true,
+        schedule: {
+          interval: '1h',
+        },
+        actions: [],
+        createdBy: null,
+        updatedBy: null,
+        createdAt: new Date(),
+        updatedAt: new Date(),
+        throttle: null,
+        notifyWhen: null,
+      },
+    });
+
+    expect(data.timeSeriesQuery).toHaveBeenCalledWith(
+      expect.objectContaining({
+        query: {
+          aggField: undefined,
+          aggType: 'foo',
+          dateEnd: '1970-01-01T00:00:00.000Z',
+          dateStart: '1970-01-01T00:00:00.000Z',
+          groupBy: 'all',
+          index: 'index-name',
+          interval: undefined,
+          termField: undefined,
+          termSize: undefined,
+          timeField: 'time-field',
+          timeWindowSize: 5,
+          timeWindowUnit: 'm',
+        },
+        condition: {
+          conditionScript: `${TIME_SERIES_BUCKET_SELECTOR_FIELD} < 1L`,
+          resultLimit: 1000,
+        },
+      })
+    );
+  });
 });

x-pack/plugins/stack_alerts/server/alert_types/index_threshold/rule_type.ts

@@ -10,21 +10,23 @@ import { Logger } from '@kbn/core/server';
 import {
   CoreQueryParamsSchemaProperties,
   TimeSeriesQuery,
+  TIME_SERIES_BUCKET_SELECTOR_FIELD,
 } from '@kbn/triggers-actions-ui-plugin/server';
 import { RuleType, RuleExecutorOptions, StackAlertsStartDeps } from '../../types';
-import { Params, ParamsSchema } from './alert_type_params';
+import { Params, ParamsSchema } from './rule_type_params';
 import { ActionContext, BaseActionContext, addMessages } from './action_context';
 import { STACK_ALERTS_FEATURE_ID } from '../../../common';
 import { ComparatorFns, getHumanReadableComparator } from '../lib';
+import { getComparatorScript } from '../lib/comparator';
 
 export const ID = '.index-threshold';
 export const ActionGroupId = 'threshold met';
 
-export function getAlertType(
+export function getRuleType(
   logger: Logger,
   data: Promise<StackAlertsStartDeps['triggersActionsUi']['data']>
 ): RuleType<Params, never, {}, {}, ActionContext, typeof ActionGroupId> {
-  const alertTypeName = i18n.translate('xpack.stackAlerts.indexThreshold.alertTypeTitle', {
+  const ruleTypeName = i18n.translate('xpack.stackAlerts.indexThreshold.alertTypeTitle', {
     defaultMessage: 'Index threshold',
   });
 
@@ -92,7 +94,7 @@ export function getAlertType(
     }
   );
 
-  const alertParamsVariables = Object.keys(CoreQueryParamsSchemaProperties).map(
+  const ruleParamsVariables = Object.keys(CoreQueryParamsSchemaProperties).map(
     (propKey: string) => {
       return {
         name: propKey,
@@ -103,7 +105,7 @@ export function getAlertType(
 
   return {
     id: ID,
-    name: alertTypeName,
+    name: ruleTypeName,
     actionGroups: [{ id: ActionGroupId, name: actionGroupName }],
     defaultActionGroupId: ActionGroupId,
     validate: {
@@ -121,7 +123,7 @@ export function getAlertType(
       params: [
         { name: 'threshold', description: actionVariableContextThresholdLabel },
         { name: 'thresholdComparator', description: actionVariableContextThresholdComparatorLabel },
-        ...alertParamsVariables,
+        ...ruleParamsVariables,
       ],
     },
     minimumLicenseRequired: 'basic',
@@ -137,6 +139,8 @@ export function getAlertType(
     const { alertId: ruleId, name, services, params } = options;
     const { alertFactory, scopedClusterClient } = services;
 
+    const alertLimit = alertFactory.alertLimit.getValue();
+
     const compareFn = ComparatorFns.get(params.thresholdComparator);
     if (compareFn == null) {
       throw new Error(
@@ -173,9 +177,19 @@ export function getAlertType(
       logger,
       esClient,
       query: queryParams,
+      condition: {
+        resultLimit: alertLimit,
+        conditionScript: getComparatorScript(
+          params.thresholdComparator,
+          params.threshold,
+          TIME_SERIES_BUCKET_SELECTOR_FIELD
+        ),
+      },
     });
     logger.debug(`rule ${ID}:${ruleId} "${name}" query result: ${JSON.stringify(result)}`);
 
+    const isGroupAgg = !!queryParams.termField;
+
     const unmetGroupValues: Record<string, number> = {};
     const agg = params.aggField ? `${params.aggType}(${params.aggField})` : `${params.aggType}`;
 
@@ -196,7 +210,10 @@ export function getAlertType(
         continue;
       }
 
-      const met = compareFn(value, params.threshold);
+      // group aggregations use the bucket selector agg to compare conditions
+      // within the ES query, so only 'met' results are returned, therefore we don't need
+      // to use the compareFn
+      const met = isGroupAgg ? true : compareFn(value, params.threshold);
 
       if (!met) {
         unmetGroupValues[alertId] = value;
@@ -219,6 +236,8 @@ export function getAlertType(
       logger.debug(`scheduled actionGroup: ${JSON.stringify(actionContext)}`);
     }
 
+    alertFactory.alertLimit.setLimitReached(result.truncated);
+
     const { getRecoveredAlerts } = services.alertFactory.done();
     for (const recoveredAlert of getRecoveredAlerts()) {
       const alertId = recoveredAlert.getId();

x-pack/plugins/stack_alerts/server/alert_types/index_threshold/rule_type_params.test.ts

@@ -5,7 +5,7 @@
  * 2.0.
  */
 
-import { ParamsSchema, Params } from './alert_type_params';
+import { ParamsSchema, Params } from './rule_type_params';
 import { ObjectType, TypeOf } from '@kbn/config-schema';
 import type { Writable } from '@kbn/utility-types';
 import { CoreQueryParams, MAX_GROUPS } from '@kbn/triggers-actions-ui-plugin/server';
@@ -22,7 +22,7 @@ const DefaultParams: Writable<Partial<Params>> = {
   threshold: [0],
 };
 
-describe('alertType Params validate()', () => {
+describe('ruleType Params validate()', () => {
   runTests(ParamsSchema, DefaultParams);
 
   // eslint-disable-next-line @typescript-eslint/no-explicit-any

x-pack/plugins/stack_alerts/server/alert_types/index_threshold/rule_type_params.ts

@@ -15,7 +15,7 @@ import { ComparatorFnNames } from '../lib';
 import { Comparator } from '../../../common/comparator_types';
 import { getComparatorSchemaType } from '../lib/comparator';
 
-// alert type parameters
+// rule type parameters
 
 export type Params = TypeOf<typeof ParamsSchema>;
 

x-pack/plugins/stack_alerts/server/alert_types/lib/comparator.test.ts

@@ -0,0 +1,47 @@
+/*
+ * 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 { getComparatorScript } from './comparator';
+import { Comparator } from '../../../common/comparator_types';
+
+describe('getComparatorScript', () => {
+  it('correctly returns script when comparator is LT', () => {
+    expect(getComparatorScript(Comparator.LT, [10], 'fieldName')).toEqual(`fieldName < 10L`);
+  });
+  it('correctly returns script when comparator is LT_OR_EQ', () => {
+    expect(getComparatorScript(Comparator.LT_OR_EQ, [10], 'fieldName')).toEqual(`fieldName <= 10L`);
+  });
+  it('correctly returns script when comparator is GT', () => {
+    expect(getComparatorScript(Comparator.GT, [10], 'fieldName')).toEqual(`fieldName > 10L`);
+  });
+  it('correctly returns script when comparator is GT_OR_EQ', () => {
+    expect(getComparatorScript(Comparator.GT_OR_EQ, [10], 'fieldName')).toEqual(`fieldName >= 10L`);
+  });
+  it('correctly returns script when comparator is BETWEEN', () => {
+    expect(getComparatorScript(Comparator.BETWEEN, [10, 100], 'fieldName')).toEqual(
+      `fieldName >= 10L && fieldName <= 100L`
+    );
+  });
+  it('correctly returns script when comparator is NOT_BETWEEN', () => {
+    expect(getComparatorScript(Comparator.NOT_BETWEEN, [10, 100], 'fieldName')).toEqual(
+      `fieldName < 10L || fieldName > 100L`
+    );
+  });
+  it('correctly returns script when threshold is float', () => {
+    expect(getComparatorScript(Comparator.LT, [3.5454], 'fieldName')).toEqual(`fieldName < 3.5454`);
+  });
+  it('throws error when threshold is empty', () => {
+    expect(() => {
+      getComparatorScript(Comparator.LT, [], 'fieldName');
+    }).toThrowErrorMatchingInlineSnapshot(`"Threshold value required"`);
+  });
+  it('throws error when comparator requires two thresholds and two thresholds are not defined', () => {
+    expect(() => {
+      getComparatorScript(Comparator.BETWEEN, [1], 'fieldName');
+    }).toThrowErrorMatchingInlineSnapshot(`"Threshold values required"`);
+  });
+});

x-pack/plugins/stack_alerts/server/alert_types/lib/comparator.ts

@@ -34,6 +34,45 @@ export const ComparatorFns = new Map<Comparator, ComparatorFn>([
   ],
 ]);
 
+export const getComparatorScript = (
+  comparator: Comparator,
+  threshold: number[],
+  fieldName: string
+) => {
+  if (threshold.length === 0) {
+    throw new Error('Threshold value required');
+  }
+
+  function getThresholdString(thresh: number) {
+    return Number.isInteger(thresh) ? `${thresh}L` : `${thresh}`;
+  }
+
+  switch (comparator) {
+    case Comparator.LT:
+      return `${fieldName} < ${getThresholdString(threshold[0])}`;
+    case Comparator.LT_OR_EQ:
+      return `${fieldName} <= ${getThresholdString(threshold[0])}`;
+    case Comparator.GT:
+      return `${fieldName} > ${getThresholdString(threshold[0])}`;
+    case Comparator.GT_OR_EQ:
+      return `${fieldName} >= ${getThresholdString(threshold[0])}`;
+    case Comparator.BETWEEN:
+      if (threshold.length < 2) {
+        throw new Error('Threshold values required');
+      }
+      return `${fieldName} >= ${getThresholdString(
+        threshold[0]
+      )} && ${fieldName} <= ${getThresholdString(threshold[1])}`;
+    case Comparator.NOT_BETWEEN:
+      if (threshold.length < 2) {
+        throw new Error('Threshold values required');
+      }
+      return `${fieldName} < ${getThresholdString(
+        threshold[0]
+      )} || ${fieldName} > ${getThresholdString(threshold[1])}`;
+  }
+};
+
 export const getComparatorSchemaType = (validate: (comparator: Comparator) => string | void) =>
   schema.oneOf(
     [

x-pack/plugins/stack_alerts/server/feature.ts

@@ -9,7 +9,7 @@ import { i18n } from '@kbn/i18n';
 import { KibanaFeatureConfig } from '@kbn/features-plugin/common';
 import { DEFAULT_APP_CATEGORIES } from '@kbn/core/server';
 import { TRANSFORM_RULE_TYPE } from '@kbn/transform-plugin/common';
-import { ID as IndexThreshold } from './alert_types/index_threshold/alert_type';
+import { ID as IndexThreshold } from './alert_types/index_threshold/rule_type';
 import { GEO_CONTAINMENT_ID as GeoContainment } from './alert_types/geo_containment/alert_type';
 import { ES_QUERY_ID as ElasticsearchQuery } from './alert_types/es_query/constants';
 import { STACK_ALERTS_FEATURE_ID } from '../common';

x-pack/plugins/stack_alerts/server/index.ts

@@ -8,7 +8,7 @@ import { get } from 'lodash';
 import { PluginConfigDescriptor, PluginInitializerContext } from '@kbn/core/server';
 import { AlertingBuiltinsPlugin } from './plugin';
 import { configSchema, Config } from '../common/config';
-export { ID as INDEX_THRESHOLD_ID } from './alert_types/index_threshold/alert_type';
+export { ID as INDEX_THRESHOLD_ID } from './alert_types/index_threshold/rule_type';
 
 export const config: PluginConfigDescriptor<Config> = {
   exposeToBrowser: {},

x-pack/plugins/triggers_actions_ui/common/data/index.ts

@@ -7,6 +7,7 @@
 
 export interface TimeSeriesResult {
   results: TimeSeriesResultRow[];
+  truncated: boolean;
 }
 
 export interface TimeSeriesResultRow {

x-pack/plugins/triggers_actions_ui/server/data/index.ts

@@ -11,6 +11,7 @@ import { registerRoutes } from './routes';
 
 export type { TimeSeriesQuery, CoreQueryParams } from './lib';
 export {
+  TIME_SERIES_BUCKET_SELECTOR_FIELD,
   CoreQueryParamsSchemaProperties,
   validateCoreQueryBody,
   validateTimeWindowUnits,

x-pack/plugins/triggers_actions_ui/server/data/lib/index.ts

@@ -6,6 +6,7 @@
  */
 
 export type { TimeSeriesQuery } from './time_series_query';
+export { TIME_SERIES_BUCKET_SELECTOR_FIELD } from './time_series_query';
 export type { CoreQueryParams } from './core_query_types';
 export {
   CoreQueryParamsSchemaProperties,

x-pack/plugins/triggers_actions_ui/server/data/lib/time_series_query.ts

@@ -12,19 +12,28 @@ import { getEsErrorMessage } from '@kbn/alerting-plugin/server';
 import { DEFAULT_GROUPS } from '..';
 import { getDateRangeInfo } from './date_range_info';
 
-import { TimeSeriesQuery, TimeSeriesResult, TimeSeriesResultRow } from './time_series_types';
+import {
+  TimeSeriesQuery,
+  TimeSeriesResult,
+  TimeSeriesResultRow,
+  TimeSeriesCondition,
+} from './time_series_types';
 export type { TimeSeriesQuery, TimeSeriesResult } from './time_series_types';
 
+export const TIME_SERIES_BUCKET_SELECTOR_PATH_NAME = 'compareValue';
+export const TIME_SERIES_BUCKET_SELECTOR_FIELD = `params.${TIME_SERIES_BUCKET_SELECTOR_PATH_NAME}`;
+
 export interface TimeSeriesQueryParameters {
   logger: Logger;
   esClient: ElasticsearchClient;
   query: TimeSeriesQuery;
+  condition?: TimeSeriesCondition;
 }
 
 export async function timeSeriesQuery(
   params: TimeSeriesQueryParameters
 ): Promise<TimeSeriesResult> {
-  const { logger, esClient, query: queryParams } = params;
+  const { logger, esClient, query: queryParams, condition: conditionParams } = params;
   const { index, timeWindowSize, timeWindowUnit, interval, timeField, dateStart, dateEnd } =
     queryParams;
 
@@ -62,6 +71,22 @@ export async function timeSeriesQuery(
 
   const isCountAgg = aggType === 'count';
   const isGroupAgg = !!termField;
+  const includeConditionInQuery = !!conditionParams;
+
+  // Cap the maximum number of terms returned to the resultLimit if defined
+  // Use resultLimit + 1 because we're using the bucket selector aggregation
+  // to apply the threshold condition to the ES query. We don't seem to be
+  // able to get the true cardinality from the bucket selector (i.e., get
+  // the number of buckets that matched the selector condition without actually
+  // retrieving the bucket data). By using resultLimit + 1, we can count the number
+  // of buckets returned and if the value is greater than resultLimit, we know that
+  // there is additional alert data that we're not returning.
+  let terms = termSize || DEFAULT_GROUPS;
+  terms = includeConditionInQuery
+    ? terms > conditionParams.resultLimit
+      ? conditionParams.resultLimit + 1
+      : terms
+    : terms;
 
   let aggParent = esQuery.body;
 
@@ -71,9 +96,18 @@ export async function timeSeriesQuery(
       groupAgg: {
         terms: {
           field: termField,
-          size: termSize || DEFAULT_GROUPS,
+          size: terms,
         },
       },
+      ...(includeConditionInQuery
+        ? {
+            groupAggCount: {
+              stats_bucket: {
+                buckets_path: 'groupAgg._count',
+              },
+            },
+          }
+        : {}),
     };
 
     // if not count add an order
@@ -82,6 +116,17 @@ export async function timeSeriesQuery(
       aggParent.aggs.groupAgg.terms.order = {
         sortValueAgg: sortOrder,
       };
+    } else if (includeConditionInQuery) {
+      aggParent.aggs.groupAgg.aggs = {
+        conditionSelector: {
+          bucket_selector: {
+            buckets_path: {
+              [TIME_SERIES_BUCKET_SELECTOR_PATH_NAME]: '_count',
+            },
+            script: conditionParams.conditionScript,
+          },
+        },
+      };
     }
 
     aggParent = aggParent.aggs.groupAgg;
@@ -89,6 +134,7 @@ export async function timeSeriesQuery(
 
   // next, add the time window aggregation
   aggParent.aggs = {
+    ...aggParent.aggs,
     dateAgg: {
       date_range: {
         field: timeField,
@@ -105,6 +151,17 @@ export async function timeSeriesQuery(
         field: aggField,
       },
     };
+
+    if (isGroupAgg && includeConditionInQuery) {
+      aggParent.aggs.conditionSelector = {
+        bucket_selector: {
+          buckets_path: {
+            [TIME_SERIES_BUCKET_SELECTOR_PATH_NAME]: 'sortValueAgg',
+          },
+          script: conditionParams.conditionScript,
+        },
+      };
+    }
   }
 
   aggParent = aggParent.aggs.dateAgg;
@@ -133,19 +190,35 @@ export async function timeSeriesQuery(
   } catch (err) {
     // console.log('time_series_query.ts error\n', JSON.stringify(err, null, 4));
     logger.warn(`${logPrefix} error: ${getEsErrorMessage(err)}`);
-    return { results: [] };
+    return { results: [], truncated: false };
   }
 
   // console.log('time_series_query.ts response\n', JSON.stringify(esResult, null, 4));
   logger.debug(`${logPrefix} result: ${JSON.stringify(esResult)}`);
-  return getResultFromEs(isCountAgg, isGroupAgg, esResult);
+  return getResultFromEs({
+    isCountAgg,
+    isGroupAgg,
+    isConditionInQuery: includeConditionInQuery,
+    esResult,
+    resultLimit: conditionParams?.resultLimit,
+  });
 }
 
-export function getResultFromEs(
-  isCountAgg: boolean,
-  isGroupAgg: boolean,
-  esResult: estypes.SearchResponse<unknown>
-): TimeSeriesResult {
+interface GetResultFromEsParams {
+  isCountAgg: boolean;
+  isGroupAgg: boolean;
+  isConditionInQuery: boolean;
+  esResult: estypes.SearchResponse<unknown>;
+  resultLimit?: number;
+}
+
+export function getResultFromEs({
+  isCountAgg,
+  isGroupAgg,
+  isConditionInQuery,
+  esResult,
+  resultLimit,
+}: GetResultFromEsParams): TimeSeriesResult {
   const aggregations = esResult?.aggregations || {};
 
   // add a fake 'all documents' group aggregation, if a group aggregation wasn't used
@@ -161,11 +234,16 @@ export function getResultFromEs(
 
   // @ts-expect-error specify aggregations type explicitly
   const groupBuckets = aggregations.groupAgg?.buckets || [];
+  // @ts-expect-error specify aggregations type explicitly
+  const numGroupsTotal = aggregations.groupAggCount?.count ?? 0;
   const result: TimeSeriesResult = {
     results: [],
+    truncated: isConditionInQuery && resultLimit ? numGroupsTotal > resultLimit : false,
   };
 
   for (const groupBucket of groupBuckets) {
+    if (resultLimit && result.results.length === resultLimit) break;
+
     const groupName: string = `${groupBucket?.key}`;
     const dateBuckets = groupBucket?.dateAgg?.buckets || [];
     const groupResult: TimeSeriesResultRow = {

x-pack/plugins/triggers_actions_ui/server/data/lib/time_series_types.ts

@@ -45,6 +45,13 @@ export const TimeSeriesQuerySchema = schema.object(
   }
 );
 
+export const TimeSeriesConditionSchema = schema.object({
+  resultLimit: schema.number(),
+  conditionScript: schema.string({ minLength: 1 }),
+});
+
+export type TimeSeriesCondition = TypeOf<typeof TimeSeriesConditionSchema>;
+
 // using direct type not allowed, circular reference, so body is typed to unknown
 function validateBody(anyParams: unknown): string | undefined {
   // validate core query parts, return if it fails validation (returning string)

x-pack/plugins/triggers_actions_ui/server/index.ts

@@ -17,6 +17,7 @@ export {
   MAX_INTERVALS,
   MAX_GROUPS,
   DEFAULT_GROUPS,
+  TIME_SERIES_BUCKET_SELECTOR_FIELD,
 } from './data';
 
 export const config: PluginConfigDescriptor<ConfigSchema> = {

x-pack/test/alerting_api_integration/common/lib/es_test_index_tool.ts

@@ -50,6 +50,9 @@ export class ESTestIndexTool {
               testedValue: {
                 type: 'long',
               },
+              testedValueFloat: {
+                type: 'float',
+              },
               testedValueUnsigned: {
                 type: 'unsigned_long',
               },

x-pack/test/alerting_api_integration/spaces_only/tests/alerting/builtin_alert_types/index_threshold/alert.ts

@@ -265,6 +265,45 @@ export default function ruleTests({ getService }: FtrProviderContext) {
       expect(inGroup2).to.be.greaterThan(0);
     });
 
+    it('runs correctly: max grouped on float', async () => {
+      await createRule({
+        name: 'never fire',
+        aggType: 'max',
+        aggField: 'testedValueFloat',
+        groupBy: 'top',
+        termField: 'group',
+        termSize: 2,
+        thresholdComparator: '<',
+        threshold: [3.235423],
+      });
+
+      await createRule({
+        name: 'always fire',
+        aggType: 'max',
+        aggField: 'testedValueFloat',
+        groupBy: 'top',
+        termField: 'group',
+        termSize: 2, // two actions will fire each interval
+        thresholdComparator: '>=',
+        threshold: [200.2354364],
+      });
+
+      // create some more documents in the first group
+      await createEsDocumentsInGroups(1);
+
+      const docs = await waitForDocs(4);
+
+      for (const doc of docs) {
+        const { name, message } = doc._source.params;
+
+        expect(name).to.be('always fire');
+
+        const messagePattern =
+          /alert 'always fire' is active for group \'group-\d\':\n\n- Value: 234.2534637451172\n- Conditions Met: max\(testedValueFloat\) is greater than or equal to 200.2354364 over 15s\n- Timestamp: \d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}.\d{3}Z/;
+        expect(message).to.match(messagePattern);
+      }
+    });
+
     it('runs correctly: max grouped on unsigned long', async () => {
       await createRule({
         name: 'never fire',

x-pack/test/alerting_api_integration/spaces_only/tests/alerting/builtin_alert_types/index_threshold/time_series_query_endpoint.ts

@@ -82,6 +82,7 @@ export default function timeSeriesQueryEndpointTests({ getService }: FtrProvider
 
       const expected = {
         results: [{ group: 'all documents', metrics: [[START_DATE_PLUS_YEAR, 0]] }],
+        truncated: false,
       };
 
       expect(await runQueryExpect(query, 200)).eql(expected);
@@ -95,6 +96,7 @@ export default function timeSeriesQueryEndpointTests({ getService }: FtrProvider
 
       const expected = {
         results: [{ group: 'all documents', metrics: [[START_DATE_MINUS_YEAR, 0]] }],
+        truncated: false,
       };
 
       expect(await runQueryExpect(query, 200)).eql(expected);
@@ -108,6 +110,7 @@ export default function timeSeriesQueryEndpointTests({ getService }: FtrProvider
 
       const expected = {
         results: [{ group: 'all documents', metrics: [[START_DATE, 6]] }],
+        truncated: false,
       };
 
       expect(await runQueryExpect(query, 200)).eql(expected);
@@ -130,6 +133,7 @@ export default function timeSeriesQueryEndpointTests({ getService }: FtrProvider
             ],
           },
         ],
+        truncated: false,
       };
 
       expect(await runQueryExpect(query, 200)).eql(expected);
@@ -154,6 +158,7 @@ export default function timeSeriesQueryEndpointTests({ getService }: FtrProvider
             ],
           },
         ],
+        truncated: false,
       };
 
       expect(await runQueryExpect(query, 200)).eql(expected);
@@ -185,6 +190,7 @@ export default function timeSeriesQueryEndpointTests({ getService }: FtrProvider
             ],
           },
         ],
+        truncated: false,
       };
 
       expect(await runQueryExpect(query, 200)).eql(expected);
@@ -220,6 +226,7 @@ export default function timeSeriesQueryEndpointTests({ getService }: FtrProvider
             ],
           },
         ],
+        truncated: false,
       };
 
       expect(await runQueryExpect(query, 200)).eql(expected);
@@ -289,6 +296,7 @@ export default function timeSeriesQueryEndpointTests({ getService }: FtrProvider
       });
       const expected = {
         results: [{ group: 'all documents', metrics: [[START_DATE, 6]] }],
+        truncated: false,
       };
       expect(await runQueryExpect(query, 200)).eql(expected);
     });
@@ -303,6 +311,7 @@ export default function timeSeriesQueryEndpointTests({ getService }: FtrProvider
       });
       const expected = {
         results: [],
+        truncated: false,
       };
       expect(await runQueryExpect(query, 200)).eql(expected);
     });

x-pack/test/alerting_api_integration/spaces_only/tests/alerting/builtin_alert_types/lib/create_test_data.ts

@@ -106,6 +106,7 @@ async function createEsDocument(
     date: new Date(epochMillis).toISOString(),
     date_epoch_millis: epochMillis,
     testedValue,
+    testedValueFloat: 234.2534643,
     testedValueUnsigned: '18446744073709551615',
     '@timestamp': new Date(epochMillis).toISOString(),
     ...(group ? { group } : {}),