[Security Solution][Detection Engine] Bubbles up more error messages from ES queries to the UI

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

@@ -337,7 +337,7 @@ export const repeatedSearchResultsWithSortId = (
   guids: string[],
   ips?: string[],
   destIps?: string[]
-) => ({
+): SignalSearchResponse => ({
   took: 10,
   timed_out: false,
   _shards: {
@@ -364,7 +364,7 @@ export const repeatedSearchResultsWithNoSortId = (
   pageSize: number,
   guids: string[],
   ips?: string[]
-) => ({
+): SignalSearchResponse => ({
   took: 10,
   timed_out: false,
   _shards: {

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

@@ -8,7 +8,7 @@ import dateMath from '@elastic/datemath';
 
 import { KibanaRequest } from '../../../../../../../src/core/server';
 import { MlPluginSetup } from '../../../../../ml/server';
-import { getAnomalies } from '../../machine_learning';
+import { AnomalyResults, getAnomalies } from '../../machine_learning';
 
 export const findMlSignals = async ({
   ml,
@@ -24,15 +24,13 @@ export const findMlSignals = async ({
   anomalyThreshold: number;
   from: string;
   to: string;
-}) => {
+}): Promise<AnomalyResults> => {
   const { mlAnomalySearch } = ml.mlSystemProvider(request);
   const params = {
     jobIds: [jobId],
     threshold: anomalyThreshold,
     earliestMs: dateMath.parse(from)?.valueOf() ?? 0,
     latestMs: dateMath.parse(to)?.valueOf() ?? 0,
   };
-  const relevantAnomalies = await getAnomalies(params, mlAnomalySearch);
-
-  return relevantAnomalies;
+  return getAnomalies(params, mlAnomalySearch);
 };

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

@@ -34,6 +34,7 @@ export const findThresholdSignals = async ({
 }: FindThresholdSignalsParams): Promise<{
   searchResult: SignalSearchResponse;
   searchDuration: string;
+  searchErrors: string[];
 }> => {
   const aggregations =
     threshold && !isEmpty(threshold.field)

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

@@ -3,56 +3,18 @@
  * or more contributor license agreements. Licensed under the Elastic License;
  * you may not use this file except in compliance with the Elastic License.
  */
-import moment from 'moment';
 
-import { AlertServices } from '../../../../../alerts/server';
-import { ListClient } from '../../../../../lists/server';
-import { RuleAlertAction } from '../../../../common/detection_engine/types';
-import { RuleTypeParams, RefreshTypes } from '../types';
-import { Logger } from '../../../../../../../src/core/server';
 import { singleSearchAfter } from './single_search_after';
 import { singleBulkCreate } from './single_bulk_create';
-import { BuildRuleMessage } from './rule_messages';
-import { SignalSearchResponse } from './types';
 import { filterEventsAgainstList } from './filter_events_with_list';
-import { ExceptionListItemSchema } from '../../../../../lists/common/schemas';
-import { getSignalTimeTuples } from './utils';
-
-interface SearchAfterAndBulkCreateParams {
-  gap: moment.Duration | null;
-  previousStartedAt: Date | null | undefined;
-  ruleParams: RuleTypeParams;
-  services: AlertServices;
-  listClient: ListClient;
-  exceptionsList: ExceptionListItemSchema[];
-  logger: Logger;
-  id: string;
-  inputIndexPattern: string[];
-  signalsIndex: string;
-  name: string;
-  actions: RuleAlertAction[];
-  createdAt: string;
-  createdBy: string;
-  updatedBy: string;
-  updatedAt: string;
-  interval: string;
-  enabled: boolean;
-  pageSize: number;
-  filter: unknown;
-  refresh: RefreshTypes;
-  tags: string[];
-  throttle: string;
-  buildRuleMessage: BuildRuleMessage;
-}
-
-export interface SearchAfterAndBulkCreateReturnType {
-  success: boolean;
-  searchAfterTimes: string[];
-  bulkCreateTimes: string[];
-  lastLookBackDate: Date | null | undefined;
-  createdSignalsCount: number;
-  errors: string[];
-}
+import {
+  createSearchAfterReturnType,
+  createSearchAfterReturnTypeFromResponse,
+  createTotalHitsFromSearchResult,
+  getSignalTimeTuples,
+  mergeReturns,
+} from './utils';
+import { SearchAfterAndBulkCreateParams, SearchAfterAndBulkCreateReturnType } from './types';
 
 // search_after through documents and re-index using bulk endpoint.
 export const searchAfterAndBulkCreate = async ({
@@ -81,14 +43,7 @@ export const searchAfterAndBulkCreate = async ({
   throttle,
   buildRuleMessage,
 }: SearchAfterAndBulkCreateParams): Promise<SearchAfterAndBulkCreateReturnType> => {
-  const toReturn: SearchAfterAndBulkCreateReturnType = {
-    success: true,
-    searchAfterTimes: [],
-    bulkCreateTimes: [],
-    lastLookBackDate: null,
-    createdSignalsCount: 0,
-    errors: [],
-  };
+  let toReturn = createSearchAfterReturnType();
 
   // sortId tells us where to start our next consecutive search_after query
   let sortId: string | undefined;
@@ -108,43 +63,43 @@ export const searchAfterAndBulkCreate = async ({
     buildRuleMessage,
   });
   logger.debug(buildRuleMessage(`totalToFromTuples: ${totalToFromTuples.length}`));
+
   while (totalToFromTuples.length > 0) {
     const tuple = totalToFromTuples.pop();
     if (tuple == null || tuple.to == null || tuple.from == null) {
       logger.error(buildRuleMessage(`[-] malformed date tuple`));
-      toReturn.success = false;
-      toReturn.errors = [...new Set([...toReturn.errors, 'malformed date tuple'])];
-      return toReturn;
+      return createSearchAfterReturnType({
+        success: false,
+        errors: ['malformed date tuple'],
+      });
     }
     signalsCreatedCount = 0;
     while (signalsCreatedCount < tuple.maxSignals) {
       try {
         logger.debug(buildRuleMessage(`sortIds: ${sortId}`));
 
         // perform search_after with optionally undefined sortId
-        const {
-          searchResult,
-          searchDuration,
-        }: { searchResult: SignalSearchResponse; searchDuration: string } = await singleSearchAfter(
-          {
-            searchAfterSortId: sortId,
-            index: inputIndexPattern,
-            from: tuple.from.toISOString(),
-            to: tuple.to.toISOString(),
-            services,
-            logger,
-            filter,
-            pageSize: tuple.maxSignals < pageSize ? Math.ceil(tuple.maxSignals) : pageSize, // maximum number of docs to receive per search result.
-            timestampOverride: ruleParams.timestampOverride,
-          }
-        );
-        toReturn.searchAfterTimes.push(searchDuration);
-
+        const { searchResult, searchDuration, searchErrors } = await singleSearchAfter({
+          searchAfterSortId: sortId,
+          index: inputIndexPattern,
+          from: tuple.from.toISOString(),
+          to: tuple.to.toISOString(),
+          services,
+          logger,
+          filter,
+          pageSize: tuple.maxSignals < pageSize ? Math.ceil(tuple.maxSignals) : pageSize, // maximum number of docs to receive per search result.
+          timestampOverride: ruleParams.timestampOverride,
+        });
+        toReturn = mergeReturns([
+          toReturn,
+          createSearchAfterReturnTypeFromResponse({ searchResult }),
+          createSearchAfterReturnType({
+            searchAfterTimes: [searchDuration],
+            errors: searchErrors,
+          }),
+        ]);
         // determine if there are any candidate signals to be processed
-        const totalHits =
-          typeof searchResult.hits.total === 'number'
-            ? searchResult.hits.total
-            : searchResult.hits.total.value;
+        const totalHits = createTotalHitsFromSearchResult({ searchResult });
         logger.debug(buildRuleMessage(`totalHits: ${totalHits}`));
         logger.debug(
           buildRuleMessage(`searchResult.hit.hits.length: ${searchResult.hits.hits.length}`)
@@ -168,17 +123,11 @@ export const searchAfterAndBulkCreate = async ({
           );
           break;
         }
-        toReturn.lastLookBackDate =
-          searchResult.hits.hits.length > 0
-            ? new Date(
-                searchResult.hits.hits[searchResult.hits.hits.length - 1]?._source['@timestamp']
-              )
-            : null;
 
         // filter out the search results that match with the values found in the list.
         // the resulting set are signals to be indexed, given they are not duplicates
         // of signals already present in the signals index.
-        const filteredEvents: SignalSearchResponse = await filterEventsAgainstList({
+        const filteredEvents = await filterEventsAgainstList({
           listClient,
           exceptionsList,
           logger,
@@ -222,19 +171,21 @@ export const searchAfterAndBulkCreate = async ({
             tags,
             throttle,
           });
-          logger.debug(buildRuleMessage(`created ${createdCount} signals`));
-          toReturn.createdSignalsCount += createdCount;
+          toReturn = mergeReturns([
+            toReturn,
+            createSearchAfterReturnType({
+              success: bulkSuccess,
+              createdSignalsCount: createdCount,
+              bulkCreateTimes: bulkDuration ? [bulkDuration] : undefined,
+              errors: bulkErrors,
+            }),
+          ]);
           signalsCreatedCount += createdCount;
+          logger.debug(buildRuleMessage(`created ${createdCount} signals`));
           logger.debug(buildRuleMessage(`signalsCreatedCount: ${signalsCreatedCount}`));
-          if (bulkDuration) {
-            toReturn.bulkCreateTimes.push(bulkDuration);
-          }
-
           logger.debug(
             buildRuleMessage(`filteredEvents.hits.hits: ${filteredEvents.hits.hits.length}`)
           );
-          toReturn.success = toReturn.success && bulkSuccess;
-          toReturn.errors = [...new Set([...toReturn.errors, ...bulkErrors])];
         }
 
         // we are guaranteed to have searchResult hits at this point
@@ -249,9 +200,13 @@ export const searchAfterAndBulkCreate = async ({
         }
       } catch (exc: unknown) {
         logger.error(buildRuleMessage(`[-] search_after and bulk threw an error ${exc}`));
-        toReturn.success = false;
-        toReturn.errors = [...new Set([...toReturn.errors, `${exc}`])];
-        return toReturn;
+        return mergeReturns([
+          toReturn,
+          createSearchAfterReturnType({
+            success: false,
+            errors: [`${exc}`],
+          }),
+        ]);
       }
     }
   }

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

@@ -18,11 +18,8 @@ import {
   sortExceptionItems,
 } from './utils';
 import { parseScheduleDates } from '../../../../common/detection_engine/parse_schedule_dates';
-import { RuleExecutorOptions } from './types';
-import {
-  searchAfterAndBulkCreate,
-  SearchAfterAndBulkCreateReturnType,
-} from './search_after_bulk_create';
+import { RuleExecutorOptions, SearchAfterAndBulkCreateReturnType } from './types';
+import { searchAfterAndBulkCreate } from './search_after_bulk_create';
 import { scheduleNotificationActions } from '../notifications/schedule_notification_actions';
 import { RuleAlertType } from '../rules/types';
 import { findMlSignals } from './find_ml_signals';
@@ -36,7 +33,17 @@ jest.mock('./rule_status_saved_objects_client');
 jest.mock('./rule_status_service');
 jest.mock('./search_after_bulk_create');
 jest.mock('./get_filter');
-jest.mock('./utils');
+jest.mock('./utils', () => {
+  const original = jest.requireActual('./utils');
+  return {
+    ...original,
+    getGapBetweenRuns: jest.fn(),
+    getGapMaxCatchupRatio: jest.fn(),
+    getListsClient: jest.fn(),
+    getExceptions: jest.fn(),
+    sortExceptionItems: jest.fn(),
+  };
+});
 jest.mock('../notifications/schedule_notification_actions');
 jest.mock('./find_ml_signals');
 jest.mock('./bulk_create_ml_signals');
@@ -383,6 +390,7 @@ describe('rules_notification_alert_type', () => {
           },
         ]);
         (findMlSignals as jest.Mock).mockResolvedValue({
+          _shards: {},
           hits: {
             hits: [],
           },
@@ -401,6 +409,7 @@ describe('rules_notification_alert_type', () => {
         payload = getPayload(ruleAlert, alertServices) as jest.Mocked<RuleExecutorOptions>;
         jobsSummaryMock.mockResolvedValue([]);
         (findMlSignals as jest.Mock).mockResolvedValue({
+          _shards: {},
           hits: {
             hits: [],
           },
@@ -409,6 +418,7 @@ describe('rules_notification_alert_type', () => {
           success: true,
           bulkCreateDuration: 0,
           createdItemsCount: 0,
+          errors: [],
         });
         await alert.executor(payload);
         expect(ruleStatusService.success).not.toHaveBeenCalled();
@@ -425,6 +435,7 @@ describe('rules_notification_alert_type', () => {
           },
         ]);
         (findMlSignals as jest.Mock).mockResolvedValue({
+          _shards: { failed: 0 },
           hits: {
             hits: [{}],
           },
@@ -433,6 +444,7 @@ describe('rules_notification_alert_type', () => {
           success: true,
           bulkCreateDuration: 1,
           createdItemsCount: 1,
+          errors: [],
         });
         await alert.executor(payload);
         expect(ruleStatusService.success).toHaveBeenCalled();
@@ -460,6 +472,7 @@ describe('rules_notification_alert_type', () => {
         });
         jobsSummaryMock.mockResolvedValue([]);
         (findMlSignals as jest.Mock).mockResolvedValue({
+          _shards: { failed: 0 },
           hits: {
             hits: [{}],
           },
@@ -468,6 +481,7 @@ describe('rules_notification_alert_type', () => {
           success: true,
           bulkCreateDuration: 1,
           createdItemsCount: 1,
+          errors: [],
         });
 
         await alert.executor(payload);