Use top_hits aggregation

elastic · Nov 5, 2020 · 0c9d5e3 · 0c9d5e3
1 parent 352a139
commit 0c9d5e3
Show file tree

Hide file tree

Showing 4 changed files with 79 additions and 46 deletions.
diff --git a/...ns/security_solution/server/lib/detection_engine/signals/bulk_create_threshold_signals.ts b/...ns/security_solution/server/lib/detection_engine/signals/bulk_create_threshold_signals.ts
@@ -14,7 +14,7 @@ import { AlertServices } from '../../../../../alerts/server';
 import { RuleAlertAction } from '../../../../common/detection_engine/types';
 import { RuleTypeParams, RefreshTypes } from '../types';
 import { singleBulkCreate, SingleBulkCreateResponse } from './single_bulk_create';
-import { SignalSearchResponse, SignalSourceHit } from './types';
+import { SignalSearchResponse, SignalSourceHit, ThresholdAggregationBucket } from './types';
 import { BuildRuleMessage } from './rule_messages';
 
 // used to generate constant Threshold Signals ID when run with the same params
@@ -125,7 +125,7 @@ const getTransformedHits = (
   startedAt: Date,
   threshold: Threshold,
   ruleId: string,
-  signalQueryFields: Record<string, string>
+  filter: unknown
 ) => {
   if (isEmpty(threshold.field)) {
     const totalResults =
@@ -138,7 +138,8 @@ const getTransformedHits = (
     const source = {
       '@timestamp': new Date().toISOString(),
       threshold_count: totalResults,
-      ...signalQueryFields,
+      // TODO: how to get signal query fields for this case???
+      // ...signalQueryFields,
     };
 
     return [
@@ -154,24 +155,30 @@ const getTransformedHits = (
     return [];
   }
 
-  return results.aggregations.threshold.buckets.map(
-    // eslint-disable-next-line @typescript-eslint/naming-convention
-    ({ key, doc_count }: { key: string; doc_count: number }) => {
-      const source = {
-        '@timestamp': new Date().toISOString(),
-        threshold_count: doc_count,
-        ...signalQueryFields,
-      };
+  return results.aggregations.threshold.buckets
+    .map(
+      ({ key, doc_count: docCount, top_threshold_hits: topHits }: ThresholdAggregationBucket) => {
+        const hit = topHits.hits.hits[0];
+        if (hit == null) {
+          return null;
+        }
 
-      set(source, threshold.field, key);
+        const source = {
+          '@timestamp': new Date().toISOString(), // TODO: use timestamp of latest event?
+          threshold_count: docCount,
+          ...getThresholdSignalQueryFields(hit, filter),
+        };
 
-      return {
-        _index: inputIndex,
-        _id: uuidv5(`${ruleId}${startedAt}${threshold.field}${key}`, NAMESPACE_ID),
-        _source: source,
-      };
-    }
-  );
+        set(source, threshold.field, key);
+
+        return {
+          _index: inputIndex,
+          _id: uuidv5(`${ruleId}${startedAt}${threshold.field}${key}`, NAMESPACE_ID),
+          _source: source,
+        };
+      }
+    )
+    .filter((bucket: ThresholdAggregationBucket) => bucket != null);
 };
 
 export const transformThresholdResultsToEcs = (
@@ -182,14 +189,13 @@ export const transformThresholdResultsToEcs = (
   threshold: Threshold,
   ruleId: string
 ): SignalSearchResponse => {
-  const signalQueryFields = getThresholdSignalQueryFields(results.hits.hits[0], filter);
   const transformedHits = getTransformedHits(
     results,
     inputIndex,
     startedAt,
     threshold,
     ruleId,
-    signalQueryFields
+    filter
   );
   const thresholdResults = {
     ...results,

diff --git a/...k/plugins/security_solution/server/lib/detection_engine/signals/find_threshold_signals.ts b/...k/plugins/security_solution/server/lib/detection_engine/signals/find_threshold_signals.ts
@@ -52,6 +52,22 @@ export const findThresholdSignals = async ({
               field: threshold.field,
               min_doc_count: threshold.value,
             },
+            aggs: {
+              // Get the most recent hit per bucket
+              top_threshold_hits: {
+                top_hits: {
+                  sort: [
+                    {
+                      '@timestamp': {
+                        // TODO: custom timestamp fields???
+                        order: 'desc',
+                      },
+                    },
+                  ],
+                  size: 1,
+                },
+              },
+            },
           },
         }
       : {};
@@ -66,7 +82,7 @@ export const findThresholdSignals = async ({
     services,
     logger,
     filter,
-    pageSize: 1,
+    pageSize: 0,
     buildRuleMessage,
   });
 };
diff --git a/x-pack/plugins/security_solution/server/lib/detection_engine/signals/types.ts b/x-pack/plugins/security_solution/server/lib/detection_engine/signals/types.ts
@@ -14,7 +14,7 @@ import {
   AlertExecutorOptions,
   AlertServices,
 } from '../../../../../alerts/server';
-import { SearchResponse } from '../../types';
+import { BaseSearchResponse, SearchResponse, TermAggregationBucket } from '../../types';
 import {
   EqlSearchResponse,
   BaseHit,
@@ -234,3 +234,7 @@ export interface SearchAfterAndBulkCreateReturnType {
   createdSignalsCount: number;
   errors: string[];
 }
+
+export interface ThresholdAggregationBucket extends TermAggregationBucket {
+  top_threshold_hits: BaseSearchResponse<SignalSource>;
+}
diff --git a/x-pack/plugins/security_solution/server/lib/types.ts b/x-pack/plugins/security_solution/server/lib/types.ts
@@ -69,41 +69,48 @@ export type ShardError = Partial<{
   }>;
 }>;
 
-export interface SearchResponse<T> {
+export interface SearchHits<T> {
+  total: TotalValue | number;
+  max_score: number;
+  hits: Array<
+    BaseHit<T> & {
+      _type: string;
+      _score: number;
+      _version?: number;
+      _explanation?: Explanation;
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      highlight?: any;
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      inner_hits?: any;
+      matched_queries?: string[];
+      sort?: string[];
+    }
+  >;
+}
+
+export interface BaseSearchResponse<T> {
+  hits: SearchHits<T>;
+}
+
+export interface SearchResponse<T> extends BaseSearchResponse<T> {
   took: number;
   timed_out: boolean;
   _scroll_id?: string;
   _shards: ShardsResponse;
-  hits: {
-    total: TotalValue | number;
-    max_score: number;
-    hits: Array<
-      BaseHit<T> & {
-        _type: string;
-        _score: number;
-        _version?: number;
-        _explanation?: Explanation;
-        // eslint-disable-next-line @typescript-eslint/no-explicit-any
-        highlight?: any;
-        // eslint-disable-next-line @typescript-eslint/no-explicit-any
-        inner_hits?: any;
-        matched_queries?: string[];
-        sort?: string[];
-      }
-    >;
-  };
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
   aggregations?: any;
 }
 
 export type SearchHit = SearchResponse<object>['hits']['hits'][0];
 
+export interface TermAggregationBucket {
+  key: string;
+  doc_count: number;
+}
+
 export interface TermAggregation {
   [agg: string]: {
-    buckets: Array<{
-      key: string;
-      doc_count: number;
-    }>;
+    buckets: TermAggregationBucket[];
   };
 }