[Security Solution][Detections] Handle dupes when processing threshold rules

x-pack/plugins/security_solution/server/lib/detection_engine/signals/find_previous_threshold_signals.ts

@@ -0,0 +1,76 @@
+/*
+ * 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 { TimestampOverrideOrUndefined } from '../../../../common/detection_engine/schemas/common/schemas';
+import { singleSearchAfter } from './single_search_after';
+
+import { AlertServices } from '../../../../../alerts/server';
+import { Logger } from '../../../../../../../src/core/server';
+import { SignalSearchResponse } from './types';
+import { BuildRuleMessage } from './rule_messages';
+
+interface FindPreviousThresholdSignalsParams {
+  from: string;
+  to: string;
+  indexPattern: string[];
+  services: AlertServices;
+  logger: Logger;
+  ruleId: string;
+  bucketByField: string;
+  timestampOverride: TimestampOverrideOrUndefined;
+  buildRuleMessage: BuildRuleMessage;
+}
+
+export const findPreviousThresholdSignals = async ({
+  from,
+  to,
+  indexPattern,
+  services,
+  logger,
+  ruleId,
+  bucketByField,
+  timestampOverride,
+  buildRuleMessage,
+}: FindPreviousThresholdSignalsParams): Promise<{
+  searchResult: SignalSearchResponse;
+  searchDuration: string;
+  searchErrors: string[];
+}> => {
+  const aggregations = {
+    threshold: {
+      terms: {
+        field: bucketByField ?? 'signal.rule.rule_id',
+      },
+      aggs: {
+        lastSignalTimestamp: {
+          max: {
+            field: 'signal.original_time', // timestamp of last event captured by bucket
+          },
+        },
+      },
+    },
+  };
+
+  const filter = {
+    term: {
+      'signal.rule.rule_id': ruleId,
+    },
+  };
+
+  return singleSearchAfter({
+    aggregations,
+    searchAfterSortId: undefined,
+    timestampOverride,
+    index: indexPattern,
+    from,
+    to,
+    services,
+    logger,
+    filter,
+    pageSize: 0,
+    buildRuleMessage,
+  });
+};

x-pack/plugins/security_solution/server/lib/detection_engine/signals/find_threshold_signals.ts

@@ -51,6 +51,7 @@ export const findThresholdSignals = async ({
             terms: {
               field: threshold.field,
               min_doc_count: threshold.value,
+              size: 10000, // max 10k buckets
             },
             aggs: {
               // Get the most recent hit per bucket

x-pack/plugins/security_solution/server/lib/detection_engine/signals/signal_rule_alert_type.ts

@@ -8,6 +8,7 @@
 
 import { Logger, KibanaRequest } from 'src/core/server';
 
+import { Filter } from 'src/plugins/data/common';
 import {
   SIGNALS_ID,
   DEFAULT_SEARCH_AFTER_PAGE_SIZE,
@@ -29,6 +30,7 @@ import {
   RuleAlertAttributes,
   EqlSignalSearchResponse,
   BaseSignalHit,
+  ThresholdQueryBucket,
 } from './types';
 import {
   getGapBetweenRuns,
@@ -46,6 +48,7 @@ import { signalParamsSchema } from './signal_params_schema';
 import { siemRuleActionGroups } from './siem_rule_action_groups';
 import { findMlSignals } from './find_ml_signals';
 import { findThresholdSignals } from './find_threshold_signals';
+import { findPreviousThresholdSignals } from './find_previous_threshold_signals';
 import { bulkCreateMlSignals } from './bulk_create_ml_signals';
 import { bulkCreateThresholdSignals } from './bulk_create_threshold_signals';
 import {
@@ -300,6 +303,48 @@ export const signalRulesAlertType = ({
             lists: exceptionItems ?? [],
           });
 
+          const {
+            searchResult: previousSignals,
+            searchErrors: previousSearchErrors,
+          } = await findPreviousThresholdSignals({
+            indexPattern: [outputIndex],
+            from,
+            to,
+            services,
+            logger,
+            ruleId,
+            bucketByField: threshold.field,
+            timestampOverride,
+            buildRuleMessage,
+          });
+
+          previousSignals.aggregations.rule.threshold.buckets.forEach(
+            (bucket: ThresholdQueryBucket) => {
+              esFilter.bool.filter.push(({
+                bool: {
+                  must_not: {
+                    bool: {
+                      must: [
+                        {
+                          term: {
+                            [threshold.field ?? 'signal.rule.rule_id']: bucket.key,
+                          },
+                        },
+                        {
+                          range: {
+                            '@timestamp': {
+                              lte: bucket.lastSignalTimestamp.value_as_string,
+                            },
+                          },
+                        },
+                      ],
+                    },
+                  },
+                },
+              } as unknown) as Filter);
+            }
+          );
+
           const { searchResult: thresholdResults, searchErrors } = await findThresholdSignals({
             inputIndexPattern: inputIndex,
             from,
@@ -349,7 +394,7 @@ export const signalRulesAlertType = ({
             }),
             createSearchAfterReturnType({
               success,
-              errors: [...errors, ...searchErrors],
+              errors: [...errors, ...previousSearchErrors, ...searchErrors],
               createdSignalsCount: createdItemsCount,
               bulkCreateTimes: bulkCreateDuration ? [bulkCreateDuration] : [],
             }),

x-pack/plugins/security_solution/server/lib/detection_engine/signals/types.ts

@@ -239,3 +239,9 @@ export interface SearchAfterAndBulkCreateReturnType {
 export interface ThresholdAggregationBucket extends TermAggregationBucket {
   top_threshold_hits: BaseSearchResponse<SignalSource>;
 }
+
+export interface ThresholdQueryBucket extends TermAggregationBucket {
+  lastSignalTimestamp: {
+    value_as_string: string;
+  };
+}