[SecuritySolution] Remove transform delay time + add unattended setti…

…ng (#184797) ## Summary * Refresh risk score index after persisting risk score * Schedule the risk score transform after calculating the risk score for a single entity * Update transform config * Add `delay: 0s` * Add `managed` and `managed_by` metadata * Add `version` metadata (used by the migration) * Add `unattended: true` * Create a transform migration ## How to test it? ### Migration 1. Install risk engine on an old version 2. Upgrade version 3. Add new alerts with new host and user 4. Run the risk engine (you can wait or force it to run by switching the flag on/off) 5. Open the explore page (user|host)/risk-tab and check if the new user|host is present ### New installation 1. Install risk engine on an empty cluster 3. Add new alerts with new host and user 4. Run the risk engine (you can wait or force it to run by switching the flag on/off) 5. Open the explore page (user|host)/risk-tab and check if the new user|host is present ### Checklist Delete any items that are not applicable to this PR. - [x] [Unit or functional tests](https://www.elastic.co/guide/en/kibana/master/development-tests.html) were updated or added to match the most common scenarios
elastic · Jun 7, 2024 · 90ae0f2 · 90ae0f2
1 parent 0ebc1a0
commit 90ae0f2
Show file tree

Hide file tree

Showing 14 changed files with 290 additions and 52 deletions.
diff --git a/...security_solution/server/lib/entity_analytics/risk_engine/risk_engine_data_client.test.ts b/...security_solution/server/lib/entity_analytics/risk_engine/risk_engine_data_client.test.ts
@@ -57,7 +57,7 @@ jest.mock('../utils/create_or_update_index', () => ({
 }));
 
 jest.spyOn(transforms, 'createTransform').mockResolvedValue(Promise.resolve());
-jest.spyOn(transforms, 'startTransform').mockResolvedValue(Promise.resolve());
+jest.spyOn(transforms, 'scheduleTransformNow').mockResolvedValue(Promise.resolve());
 
 describe('RiskEngineDataClient', () => {
   for (const useDataStreamForAlerts of [false, true]) {

diff --git a/...k/plugins/security_solution/server/lib/entity_analytics/risk_score/configurations.test.ts b/...k/plugins/security_solution/server/lib/entity_analytics/risk_score/configurations.test.ts
@@ -0,0 +1,52 @@
+/*
+ * 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 { getTransformOptions } from './configurations';
+
+describe('getTransformOptions', () => {
+  it('transform content has changed, please update the transform version and regenerate the snapshot', () => {
+    const options = getTransformOptions({
+      dest: 'dest',
+      source: ['source'],
+    });
+
+    expect(options).toMatchInlineSnapshot(`
+      Object {
+        "_meta": Object {
+          "managed": true,
+          "managed_by": "security-entity-analytics",
+          "version": 2,
+        },
+        "dest": Object {
+          "index": "dest",
+        },
+        "frequency": "1h",
+        "latest": Object {
+          "sort": "@timestamp",
+          "unique_key": Array [
+            "host.name",
+            "user.name",
+          ],
+        },
+        "settings": Object {
+          "unattended": true,
+        },
+        "source": Object {
+          "index": Array [
+            "source",
+          ],
+        },
+        "sync": Object {
+          "time": Object {
+            "delay": "0s",
+            "field": "@timestamp",
+          },
+        },
+      }
+    `);
+  });
+});
diff --git a/x-pack/plugins/security_solution/server/lib/entity_analytics/risk_score/configurations.ts b/x-pack/plugins/security_solution/server/lib/entity_analytics/risk_score/configurations.ts
@@ -4,6 +4,7 @@
  * 2.0; you may not use this file except in compliance with the Elastic License
  * 2.0.
  */
+import type { TransformPutTransformRequest } from '@elastic/elasticsearch/lib/api/types';
 import type { FieldMap } from '@kbn/alerts-as-data-utils';
 import type { IdentifierType } from '../../../../common/entity_analytics/risk_engine';
 import {
@@ -135,22 +136,45 @@ export const getIndexPatternDataStream = (namespace: string): IIndexPatternStrin
   alias: `${riskScoreBaseIndexName}.${riskScoreBaseIndexName}-${namespace}`,
 });
 
-export const getTransformOptions = ({ dest, source }: { dest: string; source: string[] }) => ({
+export type TransformOptions = Omit<TransformPutTransformRequest, 'transform_id'>;
+
+/**
+ * WARNING: We must increase the version when changing any configuration
+ *
+ * The risk engine starts the transforms executions after writing the documents to the risk score index.
+ * So the transform don't need to run on a schedule.
+ */
+export const getTransformOptions = ({
+  dest,
+  source,
+}: {
+  dest: string;
+  source: string[];
+}): Omit<TransformPutTransformRequest, 'transform_id'> => ({
   dest: {
     index: dest,
   },
-  frequency: '1h',
   latest: {
     sort: '@timestamp',
     unique_key: [`host.name`, `user.name`],
   },
   source: {
     index: source,
   },
+  frequency: '1h', // 1h is the maximum value
   sync: {
     time: {
-      delay: '2s',
+      delay: '0s', // It doesn't have any delay because the risk engine writes the documents to the index and schedules the transform synchronously.
       field: '@timestamp',
     },
   },
+  settings: {
+    unattended: true, // In unattended mode, the transform retries indefinitely in case of an error
+  },
+  _meta: {
+    version: 2, // When this field is updated we automatically update the transform
+
+    managed: true, // Metadata that identifies the transform. It has no functionality
+    managed_by: 'security-entity-analytics', // Metadata that identifies the transform. It has no functionality
+  },
 });
diff --git a/...s/security_solution/server/lib/entity_analytics/risk_score/risk_score_data_client.test.ts b/...s/security_solution/server/lib/entity_analytics/risk_score/risk_score_data_client.test.ts
@@ -36,7 +36,7 @@ jest.mock('../utils/create_or_update_index', () => ({
 }));
 
 jest.spyOn(transforms, 'createTransform').mockResolvedValue(Promise.resolve());
-jest.spyOn(transforms, 'startTransform').mockResolvedValue(Promise.resolve());
+jest.spyOn(transforms, 'scheduleTransformNow').mockResolvedValue(Promise.resolve());
 
 describe('RiskScoreDataClient', () => {
   let riskScoreDataClient: RiskScoreDataClient;
@@ -428,11 +428,19 @@ describe('RiskScoreDataClient', () => {
           },
           sync: {
             time: {
-              delay: '2s',
+              delay: '0s',
               field: '@timestamp',
             },
           },
           transform_id: 'risk_score_latest_transform_default',
+          settings: {
+            unattended: true,
+          },
+          _meta: {
+            version: 2,
+            managed: true,
+            managed_by: 'security-entity-analytics',
+          },
         },
       });
     });

diff --git a/...lugins/security_solution/server/lib/entity_analytics/risk_score/risk_score_data_client.ts b/...lugins/security_solution/server/lib/entity_analytics/risk_score/risk_score_data_client.ts
@@ -28,7 +28,10 @@ import {
 import { createDataStream } from '../utils/create_datastream';
 import type { RiskEngineDataWriter as Writer } from './risk_engine_data_writer';
 import { RiskEngineDataWriter } from './risk_engine_data_writer';
-import { getRiskScoreLatestIndex } from '../../../../common/entity_analytics/risk_engine';
+import {
+  getRiskScoreLatestIndex,
+  getRiskScoreTimeSeriesIndex,
+} from '../../../../common/entity_analytics/risk_engine';
 import { createTransform, getLatestTransformId } from '../utils/transforms';
 import { getRiskInputsIndex } from './get_risk_inputs_index';
 
@@ -71,6 +74,12 @@ export class RiskScoreDataClient {
     return writer;
   }
 
+  public refreshRiskScoreIndex = async () => {
+    await this.options.esClient.indices.refresh({
+      index: getRiskScoreTimeSeriesIndex(this.options.namespace),
+    });
+  };
+
   public getRiskInputsIndex = ({ dataViewId }: { dataViewId: string }) =>
     getRiskInputsIndex({
       dataViewId,

diff --git a/...ugins/security_solution/server/lib/entity_analytics/risk_score/risk_score_service.mock.ts b/...ugins/security_solution/server/lib/entity_analytics/risk_score/risk_score_service.mock.ts
@@ -35,6 +35,7 @@ const createRiskScoreServiceMock = (): jest.Mocked<RiskScoreService> => ({
   getConfigurationWithDefaults: jest.fn(),
   getRiskInputsIndex: jest.fn(),
   scheduleLatestTransformNow: jest.fn(),
+  refreshRiskScoreIndex: jest.fn(),
 });
 
 export const riskScoreServiceMock = {

diff --git a/...ck/plugins/security_solution/server/lib/entity_analytics/risk_score/risk_score_service.ts b/...ck/plugins/security_solution/server/lib/entity_analytics/risk_score/risk_score_service.ts
@@ -35,6 +35,7 @@ export interface RiskScoreService {
   ) => Promise<RiskEngineConfigurationWithDefaults | null>;
   getRiskInputsIndex: ({ dataViewId }: { dataViewId: string }) => Promise<RiskInputsIndexResponse>;
   scheduleLatestTransformNow: () => Promise<void>;
+  refreshRiskScoreIndex: () => Promise<void>;
 }
 
 export interface RiskScoreServiceFactoryParams {
@@ -83,5 +84,7 @@ export const riskScoreServiceFactory = ({
     };
   },
   getRiskInputsIndex: async (params) => riskScoreDataClient.getRiskInputsIndex(params),
-  scheduleLatestTransformNow: () => scheduleLatestTransformNow({ namespace: spaceId, esClient }),
+  scheduleLatestTransformNow: () =>
+    scheduleLatestTransformNow({ namespace: spaceId, esClient, logger }),
+  refreshRiskScoreIndex: () => riskScoreDataClient.refreshRiskScoreIndex(),
 });
diff --git a/...ecurity_solution/server/lib/entity_analytics/risk_score/routes/entity_calculation.test.ts b/...ecurity_solution/server/lib/entity_analytics/risk_score/routes/entity_calculation.test.ts
@@ -85,6 +85,16 @@ describe('entity risk score calculation route', () => {
     expect(response.status).toEqual(200);
   });
 
+  it('should schedule transform when risk scores are persisted ', async () => {
+    const request = buildRequest();
+
+    const response = await server.inject(request, requestContextMock.convertContext(context));
+
+    expect(mockRiskScoreService.scheduleLatestTransformNow).toHaveBeenCalled();
+
+    expect(response.status).toEqual(200);
+  });
+
   it('should call "calculateAndPersistScores" with entity filter', async () => {
     const request = buildRequest();
 

diff --git a/...ins/security_solution/server/lib/entity_analytics/risk_score/routes/entity_calculation.ts b/...ins/security_solution/server/lib/entity_analytics/risk_score/routes/entity_calculation.ts
@@ -146,6 +146,10 @@ export const riskScoreEntityCalculationRoute = (
             });
           }
 
+          if (result.scores_written > 0) {
+            await riskScoreService.scheduleLatestTransformNow();
+          }
+
           const score =
             result.scores_written === 1 ? result.scores?.[identifierType]?.[0] : undefined;
 

diff --git a/.../security_solution/server/lib/entity_analytics/risk_score/tasks/risk_scoring_task.test.ts b/.../security_solution/server/lib/entity_analytics/risk_score/tasks/risk_scoring_task.test.ts
@@ -474,19 +474,6 @@ describe('Risk Scoring Task', () => {
             expect.stringContaining('task was cancelled')
           );
         });
-
-        it('schedules the transform to run now', async () => {
-          await runTask({
-            getRiskScoreService,
-            isCancelled: mockIsCancelled,
-            logger: mockLogger,
-            taskInstance: riskScoringTaskInstanceMock,
-            telemetry: mockTelemetry,
-            entityAnalyticsConfig,
-          });
-
-          expect(mockRiskScoreService.scheduleLatestTransformNow).toHaveBeenCalledTimes(1);
-        });
       });
 
       describe('when execution was successful', () => {
@@ -520,6 +507,19 @@ describe('Risk Scoring Task', () => {
 
           expect(mockRiskScoreService.scheduleLatestTransformNow).toHaveBeenCalledTimes(1);
         });
+
+        it('refreshes the risk score index', async () => {
+          await runTask({
+            getRiskScoreService,
+            isCancelled: mockIsCancelled,
+            logger: mockLogger,
+            taskInstance: riskScoringTaskInstanceMock,
+            telemetry: mockTelemetry,
+            entityAnalyticsConfig,
+          });
+
+          expect(mockRiskScoreService.refreshRiskScoreIndex).toHaveBeenCalledTimes(1);
+        });
       });
 
       describe('when execution was unsuccessful', () => {

diff --git a/...ugins/security_solution/server/lib/entity_analytics/risk_score/tasks/risk_scoring_task.ts b/...ugins/security_solution/server/lib/entity_analytics/risk_score/tasks/risk_scoring_task.ts
@@ -310,15 +310,20 @@ export const runTask = async ({
     };
     telemetry.reportEvent(RISK_SCORE_EXECUTION_SUCCESS_EVENT.eventType, telemetryEvent);
 
-    await riskScoreService.scheduleLatestTransformNow();
-
     if (isCancelled()) {
       log('task was cancelled');
       telemetry.reportEvent(RISK_SCORE_EXECUTION_CANCELLATION_EVENT.eventType, telemetryEvent);
     }
 
+    if (scoresWritten > 0) {
+      log('refreshing risk score index and scheduling transform');
+      await riskScoreService.refreshRiskScoreIndex();
+      await riskScoreService.scheduleLatestTransformNow();
+    }
+
     log('task run completed');
     log(JSON.stringify({ ...telemetryEvent, runs }));
+
     return {
       state: updatedState,
     };

diff --git a/x-pack/plugins/security_solution/server/lib/entity_analytics/utils/transforms.test.ts b/x-pack/plugins/security_solution/server/lib/entity_analytics/utils/transforms.test.ts
@@ -5,9 +5,17 @@
  * 2.0.
  */
 
-import type { TransformGetTransformStatsResponse } from '@elastic/elasticsearch/lib/api/types';
-import { elasticsearchServiceMock } from '@kbn/core/server/mocks';
-import { scheduleTransformNow } from './transforms';
+import type {
+  TransformGetTransformResponse,
+  TransformGetTransformStatsResponse,
+} from '@elastic/elasticsearch/lib/api/types';
+import { elasticsearchServiceMock, loggingSystemMock } from '@kbn/core/server/mocks';
+import {
+  getRiskScoreLatestIndex,
+  getRiskScoreTimeSeriesIndex,
+} from '../../../../common/entity_analytics/risk_engine';
+import { getTransformOptions } from '../risk_score/configurations';
+import { scheduleLatestTransformNow, scheduleTransformNow } from './transforms';
 
 const transformId = 'test_transform_id';
 
@@ -31,6 +39,44 @@ const stoppedTransformsMock = {
   ],
 } as TransformGetTransformStatsResponse;
 
+const latestIndex = getRiskScoreLatestIndex('tests');
+const timeSeriesIndex = getRiskScoreTimeSeriesIndex('tests');
+const transformConfig = getTransformOptions({
+  dest: latestIndex,
+  source: [timeSeriesIndex],
+});
+
+const updatedTransformsMock = {
+  count: 1,
+  transforms: [
+    {
+      id: 'test_transform_id_3',
+      ...transformConfig,
+    },
+  ],
+} as TransformGetTransformResponse;
+
+const outdatedTransformsMock = {
+  count: 1,
+  transforms: [
+    {
+      ...transformConfig,
+      id: 'test_transform_id_3',
+      sync: {
+        time: {
+          field: '@timestamp',
+          delay: '2s',
+        },
+      },
+      _meta: {
+        version: '1',
+      },
+    },
+  ],
+} as TransformGetTransformResponse;
+
+const logger = loggingSystemMock.createLogger();
+
 describe('transforms utils', () => {
   beforeEach(() => {
     jest.resetAllMocks();
@@ -55,4 +101,39 @@ describe('transforms utils', () => {
       expect(esClient.transform.scheduleNowTransform).toHaveBeenCalled();
     });
   });
+
+  describe('scheduleLatestTransformNow', () => {
+    it('update the latest transform when scheduleTransformNow is called and the transform is outdated', async () => {
+      const esClient = elasticsearchServiceMock.createScopedClusterClient().asCurrentUser;
+      esClient.transform.getTransformStats.mockResolvedValueOnce(stoppedTransformsMock);
+      esClient.transform.getTransform.mockResolvedValueOnce(outdatedTransformsMock);
+
+      await scheduleLatestTransformNow({ esClient, namespace: 'tests', logger });
+
+      expect(esClient.transform.updateTransform).toHaveBeenCalled();
+    });
+
+    it('does not update the latest transform when scheduleTransformNow is called if the transform is updated', async () => {
+      const esClient = elasticsearchServiceMock.createScopedClusterClient().asCurrentUser;
+      esClient.transform.getTransformStats.mockResolvedValueOnce(stoppedTransformsMock);
+      esClient.transform.getTransform.mockResolvedValueOnce(updatedTransformsMock);
+
+      await scheduleLatestTransformNow({ esClient, namespace: 'tests', logger });
+
+      expect(esClient.transform.updateTransform).not.toHaveBeenCalled();
+    });
+
+    it('it logs the error if update transform fails', async () => {
+      const esClient = elasticsearchServiceMock.createScopedClusterClient().asCurrentUser;
+      esClient.transform.getTransformStats.mockResolvedValueOnce(stoppedTransformsMock);
+      esClient.transform.getTransform.mockResolvedValueOnce(outdatedTransformsMock);
+      esClient.transform.updateTransform.mockRejectedValueOnce(new Error('Test error'));
+
+      await scheduleLatestTransformNow({ esClient, namespace: 'tests', logger });
+
+      expect(logger.error).toHaveBeenCalledWith(
+        'There was an error upgrading the transform risk_score_latest_transform_tests. Continuing with transform scheduling. Test error'
+      );
+    });
+  });
 });