[Logs UI] [Alerting] "Group by" functionality (#68250)

- Add "group by" functionality to logs alerts

x-pack/plugins/infra/common/alerting/logs/types.ts

@@ -4,6 +4,8 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 import { i18n } from '@kbn/i18n';
+import * as rt from 'io-ts';
+import { commonSearchSuccessResponseFieldsRT } from '../../utils/elasticsearch_runtime_types';
 
 export const LOG_DOCUMENT_COUNT_ALERT_TYPE_ID = 'logs.alert.document.count';
 
@@ -20,6 +22,19 @@ export enum Comparator {
   NOT_MATCH_PHRASE = 'does not match phrase',
 }
 
+const ComparatorRT = rt.keyof({
+  [Comparator.GT]: null,
+  [Comparator.GT_OR_EQ]: null,
+  [Comparator.LT]: null,
+  [Comparator.LT_OR_EQ]: null,
+  [Comparator.EQ]: null,
+  [Comparator.NOT_EQ]: null,
+  [Comparator.MATCH]: null,
+  [Comparator.NOT_MATCH]: null,
+  [Comparator.MATCH_PHRASE]: null,
+  [Comparator.NOT_MATCH_PHRASE]: null,
+});
+
 // Maps our comparators to i18n strings, some comparators have more specific wording
 // depending on the field type the comparator is being used with.
 export const ComparatorToi18nMap = {
@@ -74,22 +89,78 @@ export enum AlertStates {
   ERROR,
 }
 
-export interface DocumentCount {
+const DocumentCountRT = rt.type({
-  comparator: Comparator;
+  comparator: ComparatorRT,
-  value: number;
+  value: rt.number,
-}
+});
 
-export interface Criterion {
+export type DocumentCount = rt.TypeOf<typeof DocumentCountRT>;
-  field: string;
-  comparator: Comparator;
-  value: string | number;
-}
 
-export interface LogDocumentCountAlertParams {
+const CriterionRT = rt.type({
-  count: DocumentCount;
+  field: rt.string,
-  criteria: Criterion[];
+  comparator: ComparatorRT,
-  timeUnit: 's' | 'm' | 'h' | 'd';
+  value: rt.union([rt.string, rt.number]),
-  timeSize: number;
+});
-}
 
-export type TimeUnit = 's' | 'm' | 'h' | 'd';
+export type Criterion = rt.TypeOf<typeof CriterionRT>;
+
+const TimeUnitRT = rt.union([rt.literal('s'), rt.literal('m'), rt.literal('h'), rt.literal('d')]);
+export type TimeUnit = rt.TypeOf<typeof TimeUnitRT>;
+
+export const LogDocumentCountAlertParamsRT = rt.intersection([
+  rt.type({
+    count: DocumentCountRT,
+    criteria: rt.array(CriterionRT),
+    timeUnit: TimeUnitRT,
+    timeSize: rt.number,
+  }),
+  rt.partial({
+    groupBy: rt.array(rt.string),
+  }),
+]);
+
+export type LogDocumentCountAlertParams = rt.TypeOf<typeof LogDocumentCountAlertParamsRT>;
+
+export const UngroupedSearchQueryResponseRT = rt.intersection([
+  commonSearchSuccessResponseFieldsRT,
+  rt.type({
+    hits: rt.type({
+      total: rt.type({
+        value: rt.number,
+      }),
+    }),
+  }),
+]);
+
+export type UngroupedSearchQueryResponse = rt.TypeOf<typeof UngroupedSearchQueryResponseRT>;
+
+export const GroupedSearchQueryResponseRT = rt.intersection([
+  commonSearchSuccessResponseFieldsRT,
+  rt.type({
+    aggregations: rt.type({
+      groups: rt.intersection([
+        rt.type({
+          buckets: rt.array(
+            rt.type({
+              key: rt.record(rt.string, rt.string),
+              doc_count: rt.number,
+              filtered_results: rt.type({
+                doc_count: rt.number,
+              }),
+            })
+          ),
+        }),
+        rt.partial({
+          after_key: rt.record(rt.string, rt.string),
+        }),
+      ]),
+    }),
+    hits: rt.type({
+      total: rt.type({
+        value: rt.number,
+      }),
+    }),
+  }),
+]);
+
+export type GroupedSearchQueryResponse = rt.TypeOf<typeof GroupedSearchQueryResponseRT>;

x-pack/plugins/infra/common/utils/elasticsearch_runtime_types.ts

@@ -0,0 +1,18 @@
+/*
+ * 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 * as rt from 'io-ts';
+
+export const commonSearchSuccessResponseFieldsRT = rt.type({
+  _shards: rt.type({
+    total: rt.number,
+    successful: rt.number,
+    skipped: rt.number,
+    failed: rt.number,
+  }),
+  timed_out: rt.boolean,
+  took: rt.number,
+});

x-pack/plugins/infra/public/components/alerting/logs/expression_editor/editor.tsx

@@ -22,6 +22,7 @@ import { DocumentCount } from './document_count';
 import { Criteria } from './criteria';
 import { useSourceId } from '../../../../containers/source_id';
 import { LogSourceProvider, useLogSourceContext } from '../../../../containers/logs/log_source';
+import { GroupByExpression } from '../../shared/group_by_expression/group_by_expression';
 
 export interface ExpressionCriteria {
   field?: string;
@@ -121,7 +122,6 @@ export const Editor: React.FC<Props> = (props) => {
   const { setAlertParams, alertParams, errors } = props;
   const [hasSetDefaults, setHasSetDefaults] = useState<boolean>(false);
   const { sourceStatus } = useLogSourceContext();
-
   useMount(() => {
     for (const [key, value] of Object.entries({ ...DEFAULT_EXPRESSION, ...alertParams })) {
       setAlertParams(key, value);
@@ -140,6 +140,17 @@ export const Editor: React.FC<Props> = (props) => {
     /* eslint-disable-next-line react-hooks/exhaustive-deps */
   }, [sourceStatus]);
 
+  const groupByFields = useMemo(() => {
+    if (sourceStatus?.logIndexFields) {
+      return sourceStatus.logIndexFields.filter((field) => {
+        return field.type === 'string' && field.aggregatable;
+      });
+    } else {
+      return [];
+    }
+    /* eslint-disable-next-line react-hooks/exhaustive-deps */
+  }, [sourceStatus]);
+
   const updateCount = useCallback(
     (countParams) => {
       const nextCountParams = { ...alertParams.count, ...countParams };
@@ -172,6 +183,13 @@ export const Editor: React.FC<Props> = (props) => {
     [setAlertParams]
   );
 
+  const updateGroupBy = useCallback(
+    (groups: string[]) => {
+      setAlertParams('groupBy', groups);
+    },
+    [setAlertParams]
+  );
+
   const addCriterion = useCallback(() => {
     const nextCriteria = alertParams?.criteria
       ? [...alertParams.criteria, DEFAULT_CRITERIA]
@@ -219,6 +237,12 @@ export const Editor: React.FC<Props> = (props) => {
         errors={errors as { [key: string]: string[] }}
       />
 
+      <GroupByExpression
+        selectedGroups={alertParams.groupBy}
+        onChange={updateGroupBy}
+        fields={groupByFields}
+      />
+
       <div>
         <EuiButtonEmpty
           color={'primary'}

x-pack/plugins/infra/public/components/alerting/logs/log_threshold_alert_type.ts

@@ -22,7 +22,7 @@ export function getAlertType(): AlertTypeModel {
     defaultActionMessage: i18n.translate(
       'xpack.infra.logs.alerting.threshold.defaultActionMessage',
       {
-        defaultMessage: `\\{\\{context.matchingDocuments\\}\\} log entries have matched the following conditions: \\{\\{context.conditions\\}\\}`,
+        defaultMessage: `\\{\\{#context.group\\}\\}\\{\\{context.group\\}\\} - \\{\\{/context.group\\}\\}\\{\\{context.matchingDocuments\\}\\} log entries have matched the following conditions: \\{\\{context.conditions\\}\\}`,
       }
     ),
     requiresAppContext: false,

x-pack/plugins/infra/public/components/alerting/shared/group_by_expression/group_by_expression.tsx

@@ -0,0 +1,85 @@
+/*
+ * 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 React, { useState, useMemo } from 'react';
+import { IFieldType } from 'src/plugins/data/public';
+import { i18n } from '@kbn/i18n';
+import {
+  EuiPopoverTitle,
+  EuiFlexItem,
+  EuiFlexGroup,
+  EuiPopover,
+  EuiExpression,
+} from '@elastic/eui';
+import { GroupBySelector } from './selector';
+
+interface Props {
+  selectedGroups?: string[];
+  fields: IFieldType[];
+  onChange: (groupBy: string[]) => void;
+  label?: string;
+}
+
+const DEFAULT_GROUP_BY_LABEL = i18n.translate('xpack.infra.alerting.alertFlyout.groupByLabel', {
+  defaultMessage: 'Group By',
+});
+
+const EVERYTHING_PLACEHOLDER = i18n.translate(
+  'xpack.infra.alerting.alertFlyout.groupBy.placeholder',
+  {
+    defaultMessage: 'Nothing (ungrouped)',
+  }
+);
+
+export const GroupByExpression: React.FC<Props> = ({
+  selectedGroups = [],
+  fields,
+  label,
+  onChange,
+}) => {
+  const [isPopoverOpen, setIsPopoverOpen] = useState(false);
+
+  const expressionValue = useMemo(() => {
+    return selectedGroups.length > 0 ? selectedGroups.join(', ') : EVERYTHING_PLACEHOLDER;
+  }, [selectedGroups]);
+
+  const labelProp = label ?? DEFAULT_GROUP_BY_LABEL;
+
+  return (
+    <EuiFlexGroup gutterSize="s">
+      <EuiFlexItem grow={false}>
+        <EuiPopover
+          id="groupByExpression"
+          button={
+            <EuiExpression
+              description={labelProp}
+              uppercase={true}
+              value={expressionValue}
+              isActive={isPopoverOpen}
+              onClick={() => setIsPopoverOpen(true)}
+            />
+          }
+          isOpen={isPopoverOpen}
+          closePopover={() => setIsPopoverOpen(false)}
+          ownFocus
+          panelPaddingSize="s"
+          anchorPosition="downLeft"
+        >
+          <div style={{ zIndex: 11000 }}>
+            <EuiPopoverTitle>{labelProp}</EuiPopoverTitle>
+            <GroupBySelector
+              selectedGroups={selectedGroups}
+              onChange={onChange}
+              fields={fields}
+              label={labelProp}
+              placeholder={EVERYTHING_PLACEHOLDER}
+            />
+          </div>
+        </EuiPopover>
+      </EuiFlexItem>
+    </EuiFlexGroup>
+  );
+};

x-pack/plugins/infra/public/components/alerting/shared/group_by_expression/selector.tsx

@@ -0,0 +1,56 @@
+/*
+ * 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 { EuiComboBox } from '@elastic/eui';
+import React, { useCallback, useMemo } from 'react';
+import { IFieldType } from 'src/plugins/data/public';
+
+interface Props {
+  selectedGroups?: string[];
+  onChange: (groupBy: string[]) => void;
+  fields: IFieldType[];
+  label: string;
+  placeholder: string;
+}
+
+export const GroupBySelector = ({
+  onChange,
+  fields,
+  selectedGroups = [],
+  label,
+  placeholder,
+}: Props) => {
+  const handleChange = useCallback(
+    (selectedOptions: Array<{ label: string }>) => {
+      const groupBy = selectedOptions.map((option) => option.label);
+      onChange(groupBy);
+    },
+    [onChange]
+  );
+
+  const formattedSelectedGroups = useMemo(() => {
+    return selectedGroups.map((group) => ({ label: group }));
+  }, [selectedGroups]);
+
+  const options = useMemo(() => {
+    return fields.filter((field) => field.aggregatable).map((field) => ({ label: field.name }));
+  }, [fields]);
+
+  return (
+    <div style={{ minWidth: '300px' }}>
+      <EuiComboBox
+        placeholder={placeholder}
+        aria-label={label}
+        fullWidth
+        singleSelection={false}
+        selectedOptions={formattedSelectedGroups}
+        options={options}
+        onChange={handleChange}
+        isClearable={true}
+      />
+    </div>
+  );
+};

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

@@ -60,6 +60,7 @@ export interface InfraDatabaseSearchResponse<Hit = {}, Aggregations = undefined>
     skipped: number;
     failed: number;
   };
+  timed_out: boolean;
   aggregations?: Aggregations;
   hits: {
     total: {

x-pack/plugins/infra/server/lib/alerting/log_threshold/log_threshold_executor.test.ts

@@ -55,7 +55,7 @@ services.alertInstanceFactory.mockImplementation((instanceId: string) => {
  * Helper functions
  */
 function getAlertState(instanceId: string): AlertStates {
-  const alert = alertInstances.get(instanceId);
+  const alert = alertInstances.get(`${instanceId}-*`);
   if (alert) {
     return alert.state.alertState;
   } else {
@@ -73,11 +73,26 @@ const executor = (createLogThresholdExecutor('test', libsMock) as unknown) as (o
 
 // Wrapper to test
 type Comparison = [number, Comparator, number];
+
 async function callExecutor(
   [value, comparator, threshold]: Comparison,
   criteria: Criterion[] = []
 ) {
-  services.callCluster.mockImplementationOnce(async (..._) => ({ count: value }));
+  services.callCluster.mockImplementationOnce(async (..._) => ({
+    _shards: {
+      total: 1,
+      successful: 1,
+      skipped: 0,
+      failed: 0,
+    },
+    timed_out: false,
+    took: 123456789,
+    hits: {
+      total: {
+        value,
+      },
+    },
+  }));
 
   return await executor({
     services,
@@ -90,222 +105,427 @@ async function callExecutor(
   });
 }
 
-describe('Comparators trigger alerts correctly', () => {
+describe('Ungrouped alerts', () => {
-  it('does not alert when counts do not reach the threshold', async () => {
+  describe('Comparators trigger alerts correctly', () => {
-    await callExecutor([0, Comparator.GT, 1]);
+    it('does not alert when counts do not reach the threshold', async () => {
-    expect(getAlertState('test')).toBe(AlertStates.OK);
+      await callExecutor([0, Comparator.GT, 1]);
+      expect(getAlertState('test')).toBe(AlertStates.OK);
 
-    await callExecutor([0, Comparator.GT_OR_EQ, 1]);
+      await callExecutor([0, Comparator.GT_OR_EQ, 1]);
-    expect(getAlertState('test')).toBe(AlertStates.OK);
+      expect(getAlertState('test')).toBe(AlertStates.OK);
 
-    await callExecutor([1, Comparator.LT, 0]);
+      await callExecutor([1, Comparator.LT, 0]);
-    expect(getAlertState('test')).toBe(AlertStates.OK);
+      expect(getAlertState('test')).toBe(AlertStates.OK);
 
-    await callExecutor([1, Comparator.LT_OR_EQ, 0]);
+      await callExecutor([1, Comparator.LT_OR_EQ, 0]);
-    expect(getAlertState('test')).toBe(AlertStates.OK);
+      expect(getAlertState('test')).toBe(AlertStates.OK);
-  });
+    });
 
-  it('alerts when counts reach the threshold', async () => {
+    it('alerts when counts reach the threshold', async () => {
-    await callExecutor([2, Comparator.GT, 1]);
+      await callExecutor([2, Comparator.GT, 1]);
-    expect(getAlertState('test')).toBe(AlertStates.ALERT);
+      expect(getAlertState('test')).toBe(AlertStates.ALERT);
 
-    await callExecutor([1, Comparator.GT_OR_EQ, 1]);
+      await callExecutor([1, Comparator.GT_OR_EQ, 1]);
-    expect(getAlertState('test')).toBe(AlertStates.ALERT);
+      expect(getAlertState('test')).toBe(AlertStates.ALERT);
 
-    await callExecutor([1, Comparator.LT, 2]);
+      await callExecutor([1, Comparator.LT, 2]);
-    expect(getAlertState('test')).toBe(AlertStates.ALERT);
+      expect(getAlertState('test')).toBe(AlertStates.ALERT);
 
-    await callExecutor([2, Comparator.LT_OR_EQ, 2]);
+      await callExecutor([2, Comparator.LT_OR_EQ, 2]);
-    expect(getAlertState('test')).toBe(AlertStates.ALERT);
+      expect(getAlertState('test')).toBe(AlertStates.ALERT);
-  });
-});
-
-describe('Comparators create the correct ES queries', () => {
-  beforeEach(() => {
-    services.callCluster.mockReset();
-  });
-
-  it('Works with `Comparator.EQ`', async () => {
-    await callExecutor(
-      [2, Comparator.GT, 1], // Not relevant
-      [{ field: 'foo', comparator: Comparator.EQ, value: 'bar' }]
-    );
-
-    const query = services.callCluster.mock.calls[0][1]!;
-    expect(query.body).toMatchObject({
-      query: {
-        bool: {
-          must: [{ term: { foo: { value: 'bar' } } }],
-        },
-      },
     });
   });
 
-  it('works with `Comparator.NOT_EQ`', async () => {
+  describe('Comparators create the correct ES queries', () => {
-    await callExecutor(
+    beforeEach(() => {
-      [2, Comparator.GT, 1], // Not relevant
+      services.callCluster.mockReset();
-      [{ field: 'foo', comparator: Comparator.NOT_EQ, value: 'bar' }]
+    });
-    );
 
-    const query = services.callCluster.mock.calls[0][1]!;
+    it('Works with `Comparator.EQ`', async () => {
-    expect(query.body).toMatchObject({
+      await callExecutor(
-      query: {
+        [2, Comparator.GT, 1], // Not relevant
-        bool: {
+        [{ field: 'foo', comparator: Comparator.EQ, value: 'bar' }]
-          must_not: [{ term: { foo: { value: 'bar' } } }],
+      );
+
+      const query = services.callCluster.mock.calls[0][1]!;
+
+      expect(query.body).toMatchObject({
+        track_total_hits: true,
+        query: {
+          bool: {
+            filter: [
+              {
+                range: {
+                  '@timestamp': {
+                    format: 'epoch_millis',
+                  },
+                },
+              },
+              {
+                term: {
+                  foo: {
+                    value: 'bar',
+                  },
+                },
+              },
+            ],
+          },
         },
-      },
+        size: 0,
+      });
+    });
+
+    it('works with `Comparator.NOT_EQ`', async () => {
+      await callExecutor(
+        [2, Comparator.GT, 1], // Not relevant
+        [{ field: 'foo', comparator: Comparator.NOT_EQ, value: 'bar' }]
+      );
+
+      const query = services.callCluster.mock.calls[0][1]!;
+
+      expect(query.body).toMatchObject({
+        track_total_hits: true,
+        query: {
+          bool: {
+            filter: [
+              {
+                range: {
+                  '@timestamp': {
+                    format: 'epoch_millis',
+                  },
+                },
+              },
+            ],
+            must_not: [
+              {
+                term: {
+                  foo: {
+                    value: 'bar',
+                  },
+                },
+              },
+            ],
+          },
+        },
+        size: 0,
+      });
+    });
+
+    it('works with `Comparator.MATCH`', async () => {
+      await callExecutor(
+        [2, Comparator.GT, 1], // Not relevant
+        [{ field: 'foo', comparator: Comparator.MATCH, value: 'bar' }]
+      );
+
+      const query = services.callCluster.mock.calls[0][1]!;
+
+      expect(query.body).toMatchObject({
+        track_total_hits: true,
+        query: {
+          bool: {
+            filter: [
+              {
+                range: {
+                  '@timestamp': {
+                    format: 'epoch_millis',
+                  },
+                },
+              },
+              {
+                match: {
+                  foo: 'bar',
+                },
+              },
+            ],
+          },
+        },
+        size: 0,
+      });
+    });
+
+    it('works with `Comparator.NOT_MATCH`', async () => {
+      await callExecutor(
+        [2, Comparator.GT, 1], // Not relevant
+        [{ field: 'foo', comparator: Comparator.NOT_MATCH, value: 'bar' }]
+      );
+
+      const query = services.callCluster.mock.calls[0][1]!;
+
+      expect(query.body).toMatchObject({
+        track_total_hits: true,
+        query: {
+          bool: {
+            filter: [
+              {
+                range: {
+                  '@timestamp': {
+                    format: 'epoch_millis',
+                  },
+                },
+              },
+            ],
+            must_not: [
+              {
+                match: {
+                  foo: 'bar',
+                },
+              },
+            ],
+          },
+        },
+        size: 0,
+      });
+    });
+
+    it('works with `Comparator.MATCH_PHRASE`', async () => {
+      await callExecutor(
+        [2, Comparator.GT, 1], // Not relevant
+        [{ field: 'foo', comparator: Comparator.MATCH_PHRASE, value: 'bar' }]
+      );
+
+      const query = services.callCluster.mock.calls[0][1]!;
+
+      expect(query.body).toMatchObject({
+        track_total_hits: true,
+        query: {
+          bool: {
+            filter: [
+              {
+                range: {
+                  '@timestamp': {
+                    format: 'epoch_millis',
+                  },
+                },
+              },
+              {
+                match_phrase: {
+                  foo: 'bar',
+                },
+              },
+            ],
+          },
+        },
+        size: 0,
+      });
+    });
+
+    it('works with `Comparator.NOT_MATCH_PHRASE`', async () => {
+      await callExecutor(
+        [2, Comparator.GT, 1], // Not relevant
+        [{ field: 'foo', comparator: Comparator.NOT_MATCH_PHRASE, value: 'bar' }]
+      );
+
+      const query = services.callCluster.mock.calls[0][1]!;
+
+      expect(query.body).toMatchObject({
+        track_total_hits: true,
+        query: {
+          bool: {
+            filter: [
+              {
+                range: {
+                  '@timestamp': {
+                    format: 'epoch_millis',
+                  },
+                },
+              },
+            ],
+            must_not: [
+              {
+                match_phrase: {
+                  foo: 'bar',
+                },
+              },
+            ],
+          },
+        },
+        size: 0,
+      });
+    });
+
+    it('works with `Comparator.GT`', async () => {
+      await callExecutor(
+        [2, Comparator.GT, 1], // Not relevant
+        [{ field: 'foo', comparator: Comparator.GT, value: 1 }]
+      );
+
+      const query = services.callCluster.mock.calls[0][1]!;
+
+      expect(query.body).toMatchObject({
+        track_total_hits: true,
+        query: {
+          bool: {
+            filter: [
+              {
+                range: {
+                  '@timestamp': {
+                    format: 'epoch_millis',
+                  },
+                },
+              },
+              {
+                range: {
+                  foo: {
+                    gt: 1,
+                  },
+                },
+              },
+            ],
+          },
+        },
+        size: 0,
+      });
+    });
+
+    it('works with `Comparator.GT_OR_EQ`', async () => {
+      await callExecutor(
+        [2, Comparator.GT, 1], // Not relevant
+        [{ field: 'foo', comparator: Comparator.GT_OR_EQ, value: 1 }]
+      );
+
+      const query = services.callCluster.mock.calls[0][1]!;
+
+      expect(query.body).toMatchObject({
+        track_total_hits: true,
+        query: {
+          bool: {
+            filter: [
+              {
+                range: {
+                  '@timestamp': {
+                    format: 'epoch_millis',
+                  },
+                },
+              },
+              {
+                range: {
+                  foo: {
+                    gte: 1,
+                  },
+                },
+              },
+            ],
+          },
+        },
+        size: 0,
+      });
+    });
+
+    it('works with `Comparator.LT`', async () => {
+      await callExecutor(
+        [2, Comparator.GT, 1], // Not relevant
+        [{ field: 'foo', comparator: Comparator.LT, value: 1 }]
+      );
+
+      const query = services.callCluster.mock.calls[0][1]!;
+
+      expect(query.body).toMatchObject({
+        track_total_hits: true,
+        query: {
+          bool: {
+            filter: [
+              {
+                range: {
+                  '@timestamp': {
+                    format: 'epoch_millis',
+                  },
+                },
+              },
+              {
+                range: {
+                  foo: {
+                    lt: 1,
+                  },
+                },
+              },
+            ],
+          },
+        },
+        size: 0,
+      });
+    });
+
+    it('works with `Comparator.LT_OR_EQ`', async () => {
+      await callExecutor(
+        [2, Comparator.GT, 1], // Not relevant
+        [{ field: 'foo', comparator: Comparator.LT_OR_EQ, value: 1 }]
+      );
+
+      const query = services.callCluster.mock.calls[0][1]!;
+
+      expect(query.body).toMatchObject({
+        track_total_hits: true,
+        query: {
+          bool: {
+            filter: [
+              {
+                range: {
+                  '@timestamp': {
+                    format: 'epoch_millis',
+                  },
+                },
+              },
+              {
+                range: {
+                  foo: {
+                    lte: 1,
+                  },
+                },
+              },
+            ],
+          },
+        },
+        size: 0,
+      });
     });
   });
 
-  it('works with `Comparator.MATCH`', async () => {
+  describe('Multiple criteria create the right ES query', () => {
-    await callExecutor(
+    beforeEach(() => {
-      [2, Comparator.GT, 1], // Not relevant
+      services.callCluster.mockReset();
-      [{ field: 'foo', comparator: Comparator.MATCH, value: 'bar' }]
-    );
-
-    const query = services.callCluster.mock.calls[0][1]!;
-    expect(query.body).toMatchObject({
-      query: {
-        bool: {
-          must: [{ match: { foo: 'bar' } }],
-        },
-      },
     });
-  });
+    it('works', async () => {
+      await callExecutor(
+        [2, Comparator.GT, 1], // Not relevant
+        [
+          { field: 'foo', comparator: Comparator.EQ, value: 'bar' },
+          { field: 'http.status', comparator: Comparator.LT, value: 400 },
+        ]
+      );
 
-  it('works with `Comparator.NOT_MATCH`', async () => {
+      const query = services.callCluster.mock.calls[0][1]!;
-    await callExecutor(
-      [2, Comparator.GT, 1], // Not relevant
-      [{ field: 'foo', comparator: Comparator.NOT_MATCH, value: 'bar' }]
-    );
 
-    const query = services.callCluster.mock.calls[0][1]!;
+      expect(query.body).toMatchObject({
-    expect(query.body).toMatchObject({
+        track_total_hits: true,
-      query: {
+        query: {
-        bool: {
+          bool: {
-          must_not: [{ match: { foo: 'bar' } }],
+            filter: [
+              {
+                range: {
+                  '@timestamp': {
+                    format: 'epoch_millis',
+                  },
+                },
+              },
+              {
+                term: {
+                  foo: {
+                    value: 'bar',
+                  },
+                },
+              },
+              {
+                range: {
+                  'http.status': {
+                    lt: 400,
+                  },
+                },
+              },
+            ],
+          },
         },
-      },
+        size: 0,
-    });
+      });
-  });
-
-  it('works with `Comparator.MATCH_PHRASE`', async () => {
-    await callExecutor(
-      [2, Comparator.GT, 1], // Not relevant
-      [{ field: 'foo', comparator: Comparator.MATCH_PHRASE, value: 'bar' }]
-    );
-
-    const query = services.callCluster.mock.calls[0][1]!;
-    expect(query.body).toMatchObject({
-      query: {
-        bool: {
-          must: [{ match_phrase: { foo: 'bar' } }],
-        },
-      },
-    });
-  });
-
-  it('works with `Comparator.NOT_MATCH_PHRASE`', async () => {
-    await callExecutor(
-      [2, Comparator.GT, 1], // Not relevant
-      [{ field: 'foo', comparator: Comparator.NOT_MATCH_PHRASE, value: 'bar' }]
-    );
-
-    const query = services.callCluster.mock.calls[0][1]!;
-    expect(query.body).toMatchObject({
-      query: {
-        bool: {
-          must_not: [{ match_phrase: { foo: 'bar' } }],
-        },
-      },
-    });
-  });
-
-  it('works with `Comparator.GT`', async () => {
-    await callExecutor(
-      [2, Comparator.GT, 1], // Not relevant
-      [{ field: 'foo', comparator: Comparator.GT, value: 1 }]
-    );
-
-    const query = services.callCluster.mock.calls[0][1]!;
-    expect(query.body).toMatchObject({
-      query: {
-        bool: {
-          must: [{ range: { foo: { gt: 1 } } }],
-        },
-      },
-    });
-  });
-
-  it('works with `Comparator.GT_OR_EQ`', async () => {
-    await callExecutor(
-      [2, Comparator.GT, 1], // Not relevant
-      [{ field: 'foo', comparator: Comparator.GT_OR_EQ, value: 1 }]
-    );
-
-    const query = services.callCluster.mock.calls[0][1]!;
-    expect(query.body).toMatchObject({
-      query: {
-        bool: {
-          must: [{ range: { foo: { gte: 1 } } }],
-        },
-      },
-    });
-  });
-
-  it('works with `Comparator.LT`', async () => {
-    await callExecutor(
-      [2, Comparator.GT, 1], // Not relevant
-      [{ field: 'foo', comparator: Comparator.LT, value: 1 }]
-    );
-
-    const query = services.callCluster.mock.calls[0][1]!;
-    expect(query.body).toMatchObject({
-      query: {
-        bool: {
-          must: [{ range: { foo: { lt: 1 } } }],
-        },
-      },
-    });
-  });
-
-  it('works with `Comparator.LT_OR_EQ`', async () => {
-    await callExecutor(
-      [2, Comparator.GT, 1], // Not relevant
-      [{ field: 'foo', comparator: Comparator.LT_OR_EQ, value: 1 }]
-    );
-
-    const query = services.callCluster.mock.calls[0][1]!;
-    expect(query.body).toMatchObject({
-      query: {
-        bool: {
-          must: [{ range: { foo: { lte: 1 } } }],
-        },
-      },
-    });
-  });
-});
-
-describe('Multiple criteria create the right ES query', () => {
-  beforeEach(() => {
-    services.callCluster.mockReset();
-  });
-  it('works', async () => {
-    await callExecutor(
-      [2, Comparator.GT, 1], // Not relevant
-      [
-        { field: 'foo', comparator: Comparator.EQ, value: 'bar' },
-        { field: 'http.status', comparator: Comparator.LT, value: 400 },
-      ]
-    );
-
-    const query = services.callCluster.mock.calls[0][1]!;
-    expect(query.body).toMatchObject({
-      query: {
-        bool: {
-          must: [{ term: { foo: { value: 'bar' } } }, { range: { 'http.status': { lt: 400 } } }],
-        },
-      },
     });
   });
 });

x-pack/plugins/infra/server/lib/alerting/log_threshold/log_threshold_executor.ts

@@ -11,10 +11,19 @@ import {
   Comparator,
   LogDocumentCountAlertParams,
   Criterion,
+  GroupedSearchQueryResponseRT,
+  UngroupedSearchQueryResponseRT,
+  UngroupedSearchQueryResponse,
+  GroupedSearchQueryResponse,
+  LogDocumentCountAlertParamsRT,
 } from '../../../../common/alerting/logs/types';
 import { InfraBackendLibs } from '../../infra_types';
 import { getIntervalInSeconds } from '../../../utils/get_interval_in_seconds';
 import { InfraSource } from '../../../../common/http_api/source_api';
+import { decodeOrThrow } from '../../../../common/runtime_types';
+
+const UNGROUPED_FACTORY_KEY = '*';
+const COMPOSITE_GROUP_SIZE = 40;
 
 const checkValueAgainstComparatorMap: {
   [key: string]: (a: number, b: number) => boolean;
@@ -25,37 +34,42 @@ const checkValueAgainstComparatorMap: {
   [Comparator.LT_OR_EQ]: (a: number, b: number) => a <= b,
 };
 
-export const createLogThresholdExecutor = (alertUUID: string, libs: InfraBackendLibs) =>
+export const createLogThresholdExecutor = (alertId: string, libs: InfraBackendLibs) =>
   async function ({ services, params }: AlertExecutorOptions) {
-    const { count, criteria } = params as LogDocumentCountAlertParams;
     const { alertInstanceFactory, savedObjectsClient, callCluster } = services;
     const { sources } = libs;
+    const { groupBy } = params;
 
     const sourceConfiguration = await sources.getSourceConfiguration(savedObjectsClient, 'default');
     const indexPattern = sourceConfiguration.configuration.logAlias;
-
+    const alertInstance = alertInstanceFactory(alertId);
-    const alertInstance = alertInstanceFactory(alertUUID);
 
     try {
-      const query = getESQuery(
+      const validatedParams = decodeOrThrow(LogDocumentCountAlertParamsRT)(params);
-        params as LogDocumentCountAlertParams,
-        sourceConfiguration.configuration
-      );
-      const result = await getResults(query, indexPattern, callCluster);
 
-      if (checkValueAgainstComparatorMap[count.comparator](result.count, count.value)) {
+      const query =
-        alertInstance.scheduleActions(FIRED_ACTIONS.id, {
+        groupBy && groupBy.length > 0
-          matchingDocuments: result.count,
+          ? getGroupedESQuery(validatedParams, sourceConfiguration.configuration, indexPattern)
-          conditions: createConditionsMessage(criteria),
+          : getUngroupedESQuery(validatedParams, sourceConfiguration.configuration, indexPattern);
-        });
 
-        alertInstance.replaceState({
+      if (!query) {
-          alertState: AlertStates.ALERT,
+        throw new Error('ES query could not be built from the provided alert params');
-        });
+      }
+
+      if (groupBy && groupBy.length > 0) {
+        processGroupByResults(
+          await getGroupedResults(query, callCluster),
+          validatedParams,
+          alertInstanceFactory,
+          alertId
+        );
       } else {
-        alertInstance.replaceState({
+        processUngroupedResults(
-          alertState: AlertStates.OK,
+          await getUngroupedResults(query, callCluster),
-        });
+          validatedParams,
+          alertInstanceFactory,
+          alertId
+        );
       }
     } catch (e) {
       alertInstance.replaceState({
@@ -66,27 +80,82 @@ export const createLogThresholdExecutor = (alertUUID: string, libs: InfraBackend
     }
   };
 
-const getESQuery = (
+const processUngroupedResults = (
+  results: UngroupedSearchQueryResponse,
   params: LogDocumentCountAlertParams,
-  sourceConfiguration: InfraSource['configuration']
+  alertInstanceFactory: AlertExecutorOptions['services']['alertInstanceFactory'],
-): object => {
+  alertId: string
+) => {
+  const { count, criteria } = params;
+
+  const alertInstance = alertInstanceFactory(`${alertId}-${UNGROUPED_FACTORY_KEY}`);
+  const documentCount = results.hits.total.value;
+
+  if (checkValueAgainstComparatorMap[count.comparator](documentCount, count.value)) {
+    alertInstance.scheduleActions(FIRED_ACTIONS.id, {
+      matchingDocuments: documentCount,
+      conditions: createConditionsMessage(criteria),
+      group: null,
+    });
+
+    alertInstance.replaceState({
+      alertState: AlertStates.ALERT,
+    });
+  } else {
+    alertInstance.replaceState({
+      alertState: AlertStates.OK,
+    });
+  }
+};
+
+interface ReducedGroupByResults {
+  name: string;
+  documentCount: number;
+}
+
+const processGroupByResults = (
+  results: GroupedSearchQueryResponse['aggregations']['groups']['buckets'],
+  params: LogDocumentCountAlertParams,
+  alertInstanceFactory: AlertExecutorOptions['services']['alertInstanceFactory'],
+  alertId: string
+) => {
+  const { count, criteria } = params;
+
+  const groupResults = results.reduce<ReducedGroupByResults[]>((acc, groupBucket) => {
+    const groupName = Object.values(groupBucket.key).join(', ');
+    const groupResult = { name: groupName, documentCount: groupBucket.filtered_results.doc_count };
+    return [...acc, groupResult];
+  }, []);
+
+  groupResults.forEach((group) => {
+    const alertInstance = alertInstanceFactory(`${alertId}-${group.name}`);
+    const documentCount = group.documentCount;
+
+    if (checkValueAgainstComparatorMap[count.comparator](documentCount, count.value)) {
+      alertInstance.scheduleActions(FIRED_ACTIONS.id, {
+        matchingDocuments: documentCount,
+        conditions: createConditionsMessage(criteria),
+        group: group.name,
+      });
+
+      alertInstance.replaceState({
+        alertState: AlertStates.ALERT,
+      });
+    } else {
+      alertInstance.replaceState({
+        alertState: AlertStates.OK,
+      });
+    }
+  });
+};
+
+const buildFiltersFromCriteria = (params: LogDocumentCountAlertParams, timestampField: string) => {
   const { timeSize, timeUnit, criteria } = params;
   const interval = `${timeSize}${timeUnit}`;
   const intervalAsSeconds = getIntervalInSeconds(interval);
+  const intervalAsMs = intervalAsSeconds * 1000;
   const to = Date.now();
-  const from = to - intervalAsSeconds * 1000;
+  const from = to - intervalAsMs;
-
-  const rangeFilters = [
-    {
-      range: {
-        [sourceConfiguration.fields.timestamp]: {
-          gte: from,
-          lte: to,
-          format: 'epoch_millis',
-        },
-      },
-    },
-  ];
 
   const positiveComparators = getPositiveComparators();
   const negativeComparators = getNegativeComparators();
@@ -101,17 +170,121 @@ const getESQuery = (
   // Negative assertions (things that "must not" match)
   const mustNotFilters = buildFiltersForCriteria(negativeCriteria);
 
-  const query = {
+  const rangeFilter = {
-    query: {
+    range: {
-      bool: {
+      [timestampField]: {
-        filter: [...rangeFilters],
+        gte: from,
-        ...(mustFilters.length > 0 && { must: mustFilters }),
+        lte: to,
-        ...(mustNotFilters.length > 0 && { must_not: mustNotFilters }),
+        format: 'epoch_millis',
       },
     },
   };
 
-  return query;
+  // For group by scenarios we'll pad the time range by 1 x the interval size on the left (lte) and right (gte), this is so
+  // a wider net is cast to "capture" the groups. This is to account for scenarios where we want ascertain if
+  // there were "no documents" (less than 1 for example). In these cases we may be missing documents to build the groups
+  // and match / not match the criteria.
+  const groupedRangeFilter = {
+    range: {
+      [timestampField]: {
+        gte: from - intervalAsMs,
+        lte: to + intervalAsMs,
+        format: 'epoch_millis',
+      },
+    },
+  };
+
+  return { rangeFilter, groupedRangeFilter, mustFilters, mustNotFilters };
+};
+
+const getGroupedESQuery = (
+  params: LogDocumentCountAlertParams,
+  sourceConfiguration: InfraSource['configuration'],
+  index: string
+): object | undefined => {
+  const { groupBy } = params;
+
+  if (!groupBy || !groupBy.length) {
+    return;
+  }
+
+  const timestampField = sourceConfiguration.fields.timestamp;
+
+  const { rangeFilter, groupedRangeFilter, mustFilters, mustNotFilters } = buildFiltersFromCriteria(
+    params,
+    timestampField
+  );
+
+  const aggregations = {
+    groups: {
+      composite: {
+        size: COMPOSITE_GROUP_SIZE,
+        sources: groupBy.map((field, groupIndex) => ({
+          [`group-${groupIndex}-${field}`]: {
+            terms: { field },
+          },
+        })),
+      },
+      aggregations: {
+        filtered_results: {
+          filter: {
+            bool: {
+              // Scope the inner filtering back to the unpadded range
+              filter: [rangeFilter, ...mustFilters],
+            },
+          },
+        },
+      },
+    },
+  };
+
+  const body = {
+    query: {
+      bool: {
+        filter: [groupedRangeFilter],
+        ...(mustNotFilters.length > 0 && { must_not: mustNotFilters }),
+      },
+    },
+    aggregations,
+    size: 0,
+  };
+
+  return {
+    index,
+    allowNoIndices: true,
+    ignoreUnavailable: true,
+    body,
+  };
+};
+
+const getUngroupedESQuery = (
+  params: LogDocumentCountAlertParams,
+  sourceConfiguration: InfraSource['configuration'],
+  index: string
+): object => {
+  const { rangeFilter, mustFilters, mustNotFilters } = buildFiltersFromCriteria(
+    params,
+    sourceConfiguration.fields.timestamp
+  );
+
+  const body = {
+    // Ensure we accurately track the hit count for the ungrouped case, otherwise we can only ensure accuracy up to 10,000.
+    track_total_hits: true,
+    query: {
+      bool: {
+        filter: [rangeFilter, ...mustFilters],
+        ...(mustNotFilters.length > 0 && { must_not: mustNotFilters }),
+      },
+    },
+    size: 0,
+  };
+
+  return {
+    index,
+    allowNoIndices: true,
+    ignoreUnavailable: true,
+    body,
+  };
 };
 
 type SupportedESQueryTypes = 'term' | 'match' | 'match_phrase' | 'range';
@@ -145,7 +318,6 @@ const buildCriterionQuery = (criterion: Criterion): Filter | undefined => {
           },
         },
       };
-      break;
     case 'match': {
       return {
         match: {
@@ -221,15 +393,31 @@ const getQueryMappingForComparator = (comparator: Comparator) => {
   return queryMappings[comparator];
 };
 
-const getResults = async (
+const getUngroupedResults = async (query: object, callCluster: AlertServices['callCluster']) => {
-  query: object,
+  return decodeOrThrow(UngroupedSearchQueryResponseRT)(await callCluster('search', query));
-  index: string,
+};
-  callCluster: AlertServices['callCluster']
+
-) => {
+const getGroupedResults = async (query: object, callCluster: AlertServices['callCluster']) => {
-  return await callCluster('count', {
+  let compositeGroupBuckets: GroupedSearchQueryResponse['aggregations']['groups']['buckets'] = [];
-    body: query,
+  let lastAfterKey: GroupedSearchQueryResponse['aggregations']['groups']['after_key'] | undefined;
-    index,
+
-  });
+  while (true) {
+    const queryWithAfterKey: any = { ...query };
+    queryWithAfterKey.body.aggregations.groups.composite.after = lastAfterKey;
+    const groupResponse: GroupedSearchQueryResponse = decodeOrThrow(GroupedSearchQueryResponseRT)(
+      await callCluster('search', queryWithAfterKey)
+    );
+    compositeGroupBuckets = [
+      ...compositeGroupBuckets,
+      ...groupResponse.aggregations.groups.buckets,
+    ];
+    lastAfterKey = groupResponse.aggregations.groups.after_key;
+    if (groupResponse.aggregations.groups.buckets.length < COMPOSITE_GROUP_SIZE) {
+      break;
+    }
+  }
+
+  return compositeGroupBuckets;
 };
 
 const createConditionsMessage = (criteria: LogDocumentCountAlertParams['criteria']) => {

x-pack/plugins/infra/server/lib/alerting/log_threshold/register_log_threshold_alert_type.ts

@@ -28,6 +28,13 @@ const conditionsActionVariableDescription = i18n.translate(
   }
 );
 
+const groupByActionVariableDescription = i18n.translate(
+  'xpack.infra.logs.alerting.threshold.groupByActionVariableDescription',
+  {
+    defaultMessage: 'The name of the group responsible for triggering the alert',
+  }
+);
+
 const countSchema = schema.object({
   value: schema.number(),
   comparator: schema.oneOf([
@@ -75,6 +82,7 @@ export async function registerLogThresholdAlertType(
         criteria: schema.arrayOf(criteriaSchema),
         timeUnit: schema.string(),
         timeSize: schema.number(),
+        groupBy: schema.maybe(schema.arrayOf(schema.string())),
       }),
     },
     defaultActionGroupId: FIRED_ACTIONS.id,
@@ -84,6 +92,7 @@ export async function registerLogThresholdAlertType(
       context: [
         { name: 'matchingDocuments', description: documentCountActionVariableDescription },
         { name: 'conditions', description: conditionsActionVariableDescription },
+        { name: 'group', description: groupByActionVariableDescription },
       ],
     },
     producer: 'logs',