[Security Solution] Refactor rules for modularization by updating bulkCreate and wrapHits methods

x-pack/plugins/security_solution/server/lib/detection_engine/notifications/rules_notification_alert_type.ts

@@ -91,7 +91,6 @@ export const rulesNotificationAlertType = ({
         signalsCount,
         resultsLink,
         ruleParams,
-        // @ts-expect-error @elastic/elasticsearch _source is optional
         signals,
       });
     }

x-pack/plugins/security_solution/server/lib/detection_engine/notifications/schedule_notification_actions.ts

@@ -8,7 +8,6 @@
 import { mapKeys, snakeCase } from 'lodash/fp';
 import { AlertInstance } from '../../../../../alerting/server';
 import { RuleParams } from '../schemas/rule_schemas';
-import { SignalSource } from '../signals/types';
 
 export type NotificationRuleTypeParams = RuleParams & {
   name: string;
@@ -20,7 +19,7 @@ interface ScheduleNotificationActions {
   signalsCount: number;
   resultsLink: string;
   ruleParams: NotificationRuleTypeParams;
-  signals: SignalSource[];
+  signals: unknown[];
 }
 
 export const scheduleNotificationActions = ({

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

@@ -0,0 +1,15 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { buildRuleMessageFactory } from '../rule_messages';
+
+export const mockBuildRuleMessage = buildRuleMessageFactory({
+  id: 'fake id',
+  ruleId: 'fake rule id',
+  index: 'fakeindex',
+  name: 'fake name',
+});

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

@@ -0,0 +1,100 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { performance } from 'perf_hooks';
+import { countBy, isEmpty, get } from 'lodash';
+import { ElasticsearchClient, Logger } from 'kibana/server';
+import { BuildRuleMessage } from './rule_messages';
+import { RefreshTypes } from '../types';
+import { BaseHit } from '../../../../common/detection_engine/types';
+import { errorAggregator, makeFloatString } from './utils';
+
+export interface GenericBulkCreateResponse<T> {
+  success: boolean;
+  bulkCreateDuration: string;
+  createdItemsCount: number;
+  createdItems: Array<T & { _id: string; _index: string }>;
+  errors: string[];
+}
+
+export const bulkCreateFactory = (
+  logger: Logger,
+  esClient: ElasticsearchClient,
+  buildRuleMessage: BuildRuleMessage,
+  refreshForBulkCreate: RefreshTypes
+) => async <T>(wrappedDocs: Array<BaseHit<T>>): Promise<GenericBulkCreateResponse<T>> => {
+  if (wrappedDocs.length === 0) {
+    return {
+      errors: [],
+      success: true,
+      bulkCreateDuration: '0',
+      createdItemsCount: 0,
+      createdItems: [],
+    };
+  }
+
+  const bulkBody = wrappedDocs.flatMap((wrappedDoc) => [
+    {
+      create: {
+        _index: wrappedDoc._index,
+        _id: wrappedDoc._id,
+      },
+    },
+    wrappedDoc._source,
+  ]);
+  const start = performance.now();
+
+  const { body: response } = await esClient.bulk({
+    refresh: refreshForBulkCreate,
+    body: bulkBody,
+  });
+
+  const end = performance.now();
+  logger.debug(
+    buildRuleMessage(
+      `individual bulk process time took: ${makeFloatString(end - start)} milliseconds`
+    )
+  );
+  logger.debug(buildRuleMessage(`took property says bulk took: ${response.took} milliseconds`));
+  const createdItems = wrappedDocs
+    .map((doc, index) => ({
+      _id: response.items[index].create?._id ?? '',
+      _index: response.items[index].create?._index ?? '',
+      ...doc._source,
+    }))
+    .filter((_, index) => get(response.items[index], 'create.status') === 201);
+  const createdItemsCount = createdItems.length;
+  const duplicateSignalsCount = countBy(response.items, 'create.status')['409'];
+  const errorCountByMessage = errorAggregator(response, [409]);
+
+  logger.debug(buildRuleMessage(`bulk created ${createdItemsCount} signals`));
+  if (duplicateSignalsCount > 0) {
+    logger.debug(buildRuleMessage(`ignored ${duplicateSignalsCount} duplicate signals`));
+  }
+  if (!isEmpty(errorCountByMessage)) {
+    logger.error(
+      buildRuleMessage(
+        `[-] bulkResponse had errors with responses of: ${JSON.stringify(errorCountByMessage)}`
+      )
+    );
+    return {
+      errors: Object.keys(errorCountByMessage),
+      success: false,
+      bulkCreateDuration: makeFloatString(end - start),
+      createdItemsCount,
+      createdItems,
+    };
+  } else {
+    return {
+      errors: [],
+      success: true,
+      bulkCreateDuration: makeFloatString(end - start),
+      createdItemsCount,
+      createdItems,
+    };
+  }
+};

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

@@ -14,11 +14,10 @@ import {
   AlertInstanceState,
   AlertServices,
 } from '../../../../../alerting/server';
-import { RefreshTypes } from '../types';
-import { singleBulkCreate, SingleBulkCreateResponse } from './single_bulk_create';
+import { GenericBulkCreateResponse } from './bulk_create_factory';
 import { AnomalyResults, Anomaly } from '../../machine_learning';
 import { BuildRuleMessage } from './rule_messages';
-import { AlertAttributes } from './types';
+import { AlertAttributes, BulkCreate, WrapHits } from './types';
 import { MachineLearningRuleParams } from '../schemas/rule_schemas';
 
 interface BulkCreateMlSignalsParams {
@@ -28,8 +27,9 @@ interface BulkCreateMlSignalsParams {
   logger: Logger;
   id: string;
   signalsIndex: string;
-  refresh: RefreshTypes;
   buildRuleMessage: BuildRuleMessage;
+  bulkCreate: BulkCreate;
+  wrapHits: WrapHits;
 }
 
 interface EcsAnomaly extends Anomaly {
@@ -85,9 +85,10 @@ const transformAnomalyResultsToEcs = (
 
 export const bulkCreateMlSignals = async (
   params: BulkCreateMlSignalsParams
-): Promise<SingleBulkCreateResponse> => {
+): Promise<GenericBulkCreateResponse<{}>> => {
   const anomalyResults = params.someResult;
   const ecsResults = transformAnomalyResultsToEcs(anomalyResults);
-  const buildRuleMessage = params.buildRuleMessage;
-  return singleBulkCreate({ ...params, filteredEvents: ecsResults, buildRuleMessage });
+
+  const wrappedDocs = params.wrapHits(ecsResults.hits.hits);
+  return params.bulkCreate(wrappedDocs);
 };

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

@@ -7,7 +7,6 @@
 
 import { loggingSystemMock } from 'src/core/server/mocks';
 import { alertsMock, AlertServicesMock } from '../../../../../../alerting/server/mocks';
-import { RuleStatusService } from '../rule_status_service';
 import { eqlExecutor } from './eql';
 import { getExceptionListItemSchemaMock } from '../../../../../../lists/common/schemas/response/exception_list_item_schema.mock';
 import { getEntryListMock } from '../../../../../../lists/common/schemas/types/entry_list.mock';
@@ -23,7 +22,6 @@ describe('eql_executor', () => {
   const version = '8.0.0';
   let logger: ReturnType<typeof loggingSystemMock.createLogger>;
   let alertServices: AlertServicesMock;
-  let ruleStatusService: Record<string, jest.Mock>;
   (getIndexVersion as jest.Mock).mockReturnValue(SIGNALS_TEMPLATE_VERSION);
   const eqlSO = {
     id: '04128c15-0d1b-4716-a4c5-46997ac7f3bd',
@@ -51,17 +49,11 @@ describe('eql_executor', () => {
   beforeEach(() => {
     alertServices = alertsMock.createAlertServices();
     logger = loggingSystemMock.createLogger();
-    ruleStatusService = {
-      success: jest.fn(),
-      find: jest.fn(),
-      goingToRun: jest.fn(),
-      error: jest.fn(),
-      partialFailure: jest.fn(),
-    };
     alertServices.scopedClusterClient.asCurrentUser.transport.request.mockResolvedValue(
       elasticsearchClientMock.createSuccessTransportRequestPromise({
         hits: {
           total: { value: 10 },
+          events: [],
         },
       })
     );
@@ -70,25 +62,16 @@ describe('eql_executor', () => {
   describe('eqlExecutor', () => {
     it('should set a warning when exception list for eql rule contains value list exceptions', async () => {
       const exceptionItems = [getExceptionListItemSchemaMock({ entries: [getEntryListMock()] })];
-      try {
-        await eqlExecutor({
-          rule: eqlSO,
-          exceptionItems,
-          ruleStatusService: (ruleStatusService as unknown) as RuleStatusService,
-          services: alertServices,
-          version,
-          logger,
-          refresh: false,
-          searchAfterSize,
-        });
-      } catch (err) {
-        // eqlExecutor will throw until we have an EQL response mock that conforms to the
-        // expected EQL response format, so just catch the error and check the status service
-      }
-      expect(ruleStatusService.partialFailure).toHaveBeenCalled();
-      expect(ruleStatusService.partialFailure.mock.calls[0][0]).toContain(
-        'Exceptions that use "is in list" or "is not in list" operators are not applied to EQL rules'
-      );
+      const response = await eqlExecutor({
+        rule: eqlSO,
+        exceptionItems,
+        services: alertServices,
+        version,
+        logger,
+        searchAfterSize,
+        bulkCreate: jest.fn(),
+      });
+      expect(response.warningMessages.length).toEqual(1);
     });
   });
 });

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

@@ -7,9 +7,9 @@
 
 import { ApiResponse } from '@elastic/elasticsearch';
 import { performance } from 'perf_hooks';
-import { Logger } from 'src/core/server';
 import { SavedObject } from 'src/core/types';
 import type { ExceptionListItemSchema } from '@kbn/securitysolution-io-ts-list-types';
+import { Logger } from 'src/core/server';
 import {
   AlertInstanceContext,
   AlertInstanceState,
@@ -21,13 +21,12 @@ import { isOutdated } from '../../migrations/helpers';
 import { getIndexVersion } from '../../routes/index/get_index_version';
 import { MIN_EQL_RULE_INDEX_VERSION } from '../../routes/index/get_signals_template';
 import { EqlRuleParams } from '../../schemas/rule_schemas';
-import { RefreshTypes } from '../../types';
 import { buildSignalFromEvent, buildSignalGroupFromSequence } from '../build_bulk_body';
 import { getInputIndex } from '../get_input_output_index';
-import { RuleStatusService } from '../rule_status_service';
-import { bulkInsertSignals, filterDuplicateSignals } from '../single_bulk_create';
+import { filterDuplicateSignals } from '../filter_duplicate_signals';
 import {
   AlertAttributes,
+  BulkCreate,
   EqlSignalSearchResponse,
   SearchAfterAndBulkCreateReturnType,
   WrappedSignalHit,
@@ -37,26 +36,24 @@ import { createSearchAfterReturnType, makeFloatString, wrapSignal } from '../uti
 export const eqlExecutor = async ({
   rule,
   exceptionItems,
-  ruleStatusService,
   services,
   version,
-  searchAfterSize,
   logger,
-  refresh,
+  searchAfterSize,
+  bulkCreate,
 }: {
   rule: SavedObject<AlertAttributes<EqlRuleParams>>;
   exceptionItems: ExceptionListItemSchema[];
-  ruleStatusService: RuleStatusService;
   services: AlertServices<AlertInstanceState, AlertInstanceContext, 'default'>;
   version: string;
-  searchAfterSize: number;
   logger: Logger;
-  refresh: RefreshTypes;
+  searchAfterSize: number;
+  bulkCreate: BulkCreate;
 }): Promise<SearchAfterAndBulkCreateReturnType> => {
   const result = createSearchAfterReturnType();
   const ruleParams = rule.attributes.params;
   if (hasLargeValueItem(exceptionItems)) {
-    await ruleStatusService.partialFailure(
+    result.warningMessages.push(
       'Exceptions that use "is in list" or "is not in list" operators are not applied to EQL rules'
     );
     result.warning = true;
@@ -125,7 +122,7 @@ export const eqlExecutor = async ({
   }
 
   if (newSignals.length > 0) {
-    const insertResult = await bulkInsertSignals(newSignals, logger, services, refresh);
+    const insertResult = await bulkCreate(newSignals);
     result.bulkCreateTimes.push(insertResult.bulkCreateDuration);
     result.createdSignalsCount += insertResult.createdItemsCount;
     result.createdSignals = insertResult.createdItems;