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

x-pack/plugins/security_solution/server/lib/detection_engine/routes/index/get_signals_template.ts

@@ -7,7 +7,7 @@
 import signalsMapping from './signals_mapping.json';
 import ecsMapping from './ecs_mapping.json';
 
-export const SIGNALS_TEMPLATE_VERSION = 2;
+export const SIGNALS_TEMPLATE_VERSION = 3;
 export const MIN_EQL_RULE_INDEX_VERSION = 2;
 
 export const getSignalsTemplate = (index: string) => {

x-pack/plugins/security_solution/server/lib/detection_engine/routes/index/signals_mapping.json

@@ -337,6 +337,16 @@
           "threshold_count": {
             "type": "float"
           },
+          "threshold_result": {
+            "properties": {
+              "count": {
+                "type": "long"
+              },
+              "value": {
+                "type": "keyword"
+              }
+            }
+          },
           "depth": {
             "type": "integer"
           }

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

@@ -123,6 +123,115 @@ describe('buildBulkBody', () => {
     expect(fakeSignalSourceHit).toEqual(expected);
   });
 
+  test('bulk body builds well-defined body with threshold results', () => {
+    const sampleParams = sampleRuleAlertParams();
+    const baseDoc = sampleDocNoSortId();
+    const doc: SignalSourceHit = {
+      ...baseDoc,
+      _source: {
+        ...baseDoc._source,
+        threshold_result: {
+          count: 5,
+          value: 'abcd',
+        },
+      },
+    };
+    delete doc._source.source;
+    const fakeSignalSourceHit = buildBulkBody({
+      doc,
+      ruleParams: sampleParams,
+      id: sampleRuleGuid,
+      name: 'rule-name',
+      actions: [],
+      createdAt: '2020-01-28T15:58:34.810Z',
+      updatedAt: '2020-01-28T15:59:14.004Z',
+      createdBy: 'elastic',
+      updatedBy: 'elastic',
+      interval: '5m',
+      enabled: true,
+      tags: ['some fake tag 1', 'some fake tag 2'],
+      throttle: 'no_actions',
+    });
+    // Timestamp will potentially always be different so remove it for the test
+    // @ts-expect-error
+    delete fakeSignalSourceHit['@timestamp'];
+    const expected: Omit<SignalHit, '@timestamp'> & { someKey: 'someValue' } = {
+      someKey: 'someValue',
+      event: {
+        kind: 'signal',
+      },
+      signal: {
+        parent: {
+          id: sampleIdGuid,
+          type: 'event',
+          index: 'myFakeSignalIndex',
+          depth: 0,
+        },
+        parents: [
+          {
+            id: sampleIdGuid,
+            type: 'event',
+            index: 'myFakeSignalIndex',
+            depth: 0,
+          },
+        ],
+        ancestors: [
+          {
+            id: sampleIdGuid,
+            type: 'event',
+            index: 'myFakeSignalIndex',
+            depth: 0,
+          },
+        ],
+        original_time: '2020-04-20T21:27:45+0000',
+        status: 'open',
+        rule: {
+          actions: [],
+          author: ['Elastic'],
+          building_block_type: 'default',
+          id: '04128c15-0d1b-4716-a4c5-46997ac7f3bd',
+          rule_id: 'rule-1',
+          false_positives: [],
+          max_signals: 10000,
+          risk_score: 50,
+          risk_score_mapping: [],
+          output_index: '.siem-signals',
+          description: 'Detecting root and admin users',
+          from: 'now-6m',
+          immutable: false,
+          index: ['auditbeat-*', 'filebeat-*', 'packetbeat-*', 'winlogbeat-*'],
+          interval: '5m',
+          language: 'kuery',
+          license: 'Elastic License',
+          name: 'rule-name',
+          query: 'user.name: root or user.name: admin',
+          references: ['http://google.com'],
+          severity: 'high',
+          severity_mapping: [],
+          tags: ['some fake tag 1', 'some fake tag 2'],
+          threat: [],
+          throttle: 'no_actions',
+          type: 'query',
+          to: 'now',
+          note: '',
+          enabled: true,
+          created_by: 'elastic',
+          updated_by: 'elastic',
+          version: 1,
+          created_at: fakeSignalSourceHit.signal.rule?.created_at,
+          updated_at: fakeSignalSourceHit.signal.rule?.updated_at,
+          exceptions_list: getListArrayMock(),
+        },
+        threshold_result: {
+          count: 5,
+          value: 'abcd',
+        },
+        depth: 1,
+      },
+    };
+    expect(fakeSignalSourceHit).toEqual(expected);
+  });
+
   test('bulk body builds original_event if it exists on the event to begin with', () => {
     const sampleParams = sampleRuleAlertParams();
     const doc = sampleDocNoSortId();

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

@@ -71,6 +71,7 @@ export const buildBulkBody = ({
     ...buildSignal([doc], rule),
     ...additionalSignalFields(doc),
   };
+  delete doc._source.threshold_result;
   const event = buildEventTypeSignal(doc);
   const signalHit: SignalHit = {
     ...doc._source,

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

@@ -96,9 +96,9 @@ export const buildSignal = (docs: BaseSignalHit[], rule: RulesSchema): Signal =>
 export const additionalSignalFields = (doc: BaseSignalHit) => {
   return {
     parent: buildParent(removeClashes(doc)),
-    original_time: doc._source['@timestamp'],
+    original_time: doc._source['@timestamp'], // This field has already been replaced with timestampOverride, if provided.
     original_event: doc._source.event ?? undefined,
-    threshold_count: doc._source.threshold_count ?? undefined,
+    threshold_result: doc._source.threshold_result,
     original_signal:
       doc._source.signal != null && !isEventTypeSignal(doc) ? doc._source.signal : undefined,
   };

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

@@ -149,7 +149,10 @@ const getTransformedHits = (
 
     const source = {
       '@timestamp': get(timestampOverride ?? '@timestamp', hit._source),
-      threshold_count: totalResults,
+      threshold_result: {
+        count: totalResults,
+        value: ruleId,
+      },
       ...getThresholdSignalQueryFields(hit, filter),
     };
 
@@ -176,7 +179,10 @@ const getTransformedHits = (
 
         const source = {
           '@timestamp': get(timestampOverride ?? '@timestamp', hit._source),
-          threshold_count: docCount,
+          threshold_result: {
+            count: docCount,
+            value: get(threshold.field, hit._source),
+          },
           ...getThresholdSignalQueryFields(hit, filter),
         };
 

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

@@ -0,0 +1,87 @@
+/*
+ * 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: 'signal.threshold_result.value',
+      },
+      aggs: {
+        lastSignalTimestamp: {
+          max: {
+            field: 'signal.original_time', // timestamp of last event captured by bucket
+          },
+        },
+      },
+    },
+  };
+
+  const filter = {
+    bool: {
+      must: [
+        {
+          term: {
+            'signal.rule.rule_id': ruleId,
+          },
+        },
+        {
+          term: {
+            'signal.rule.threshold.field': bucketByField,
+          },
+        },
+      ],
+    },
+  };
+
+  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,46 @@ 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.threshold.buckets.forEach((bucket: ThresholdQueryBucket) => {
+            esFilter.bool.filter.push(({
+              bool: {
+                must_not: {
+                  bool: {
+                    must: [
+                      {
+                        term: {
+                          [threshold.field ?? 'signal.rule.rule_id']: bucket.key,
+                        },
+                      },
+                      {
+                        range: {
+                          [timestampOverride ?? '@timestamp']: {
+                            lte: bucket.lastSignalTimestamp.value_as_string,
+                          },
+                        },
+                      },
+                    ],
+                  },
+                },
+              },
+            } as unknown) as Filter);
+          });
+
           const { searchResult: thresholdResults, searchErrors } = await findThresholdSignals({
             inputIndexPattern: inputIndex,
             from,
@@ -349,7 +392,7 @@ export const signalRulesAlertType = ({
             }),
             createSearchAfterReturnType({
               success,
-              errors: [...errors, ...searchErrors],
+              errors: [...errors, ...previousSearchErrors, ...searchErrors],
               createdSignalsCount: createdItemsCount,
               bulkCreateTimes: bulkCreateDuration ? [bulkCreateDuration] : [],
             }),