[Security Solution] create task for auto restarting failed OLM transforms (#113686) (#113770)

Co-authored-by: Joey F. Poon <joey.poon@elastic.co>

x-pack/plugins/security_solution/common/constants.ts

@@ -7,7 +7,7 @@
 
 import type { TransformConfigSchema } from './transforms/types';
 import { ENABLE_CASE_CONNECTOR } from '../../cases/common';
-import { metadataTransformPattern } from './endpoint/constants';
+import { METADATA_TRANSFORMS_PATTERN } from './endpoint/constants';
 
 export const APP_ID = 'securitySolution';
 export const SERVER_APP_ID = 'siem';
@@ -330,6 +330,23 @@ export const showAllOthersBucket: string[] = [
  */
 export const ELASTIC_NAME = 'estc';
 
-export const TRANSFORM_STATS_URL = `/api/transform/transforms/${metadataTransformPattern}-*/_stats`;
+export const METADATA_TRANSFORM_STATS_URL = `/api/transform/transforms/${METADATA_TRANSFORMS_PATTERN}/_stats`;
 
 export const RISKY_HOSTS_INDEX = 'ml_host_risk_score_latest';
+
+export const TRANSFORM_STATES = {
+  ABORTING: 'aborting',
+  FAILED: 'failed',
+  INDEXING: 'indexing',
+  STARTED: 'started',
+  STOPPED: 'stopped',
+  STOPPING: 'stopping',
+  WAITING: 'waiting',
+};
+
+export const WARNING_TRANSFORM_STATES = new Set([
+  TRANSFORM_STATES.ABORTING,
+  TRANSFORM_STATES.FAILED,
+  TRANSFORM_STATES.STOPPED,
+  TRANSFORM_STATES.STOPPING,
+]);

x-pack/plugins/security_solution/common/endpoint/constants.ts

@@ -17,10 +17,13 @@ export const metadataCurrentIndexPattern = 'metrics-endpoint.metadata_current_*'
 /** The metadata Transform Name prefix with NO (package) version) */
 export const metadataTransformPrefix = 'endpoint.metadata_current-default';
 
-/** The metadata Transform Name prefix with NO namespace and NO (package) version) */
-export const metadataTransformPattern = 'endpoint.metadata_current-*';
+// metadata transforms pattern for matching all metadata transform ids
+export const METADATA_TRANSFORMS_PATTERN = 'endpoint.metadata_*';
 
+// united metadata transform id
 export const METADATA_UNITED_TRANSFORM = 'endpoint.metadata_united-default';
+
+// united metadata transform destination index
 export const METADATA_UNITED_INDEX = '.metrics-endpoint.metadata_united_default';
 
 export const policyIndexPattern = 'metrics-endpoint.policy-*';

x-pack/plugins/security_solution/public/management/pages/endpoint_hosts/mocks.ts

@@ -37,8 +37,8 @@ import {
   PendingActionsHttpMockInterface,
   pendingActionsHttpMock,
 } from '../../../common/lib/endpoint_pending_actions/mocks';
-import { TRANSFORM_STATS_URL } from '../../../../common/constants';
-import { TransformStatsResponse, TRANSFORM_STATE } from './types';
+import { METADATA_TRANSFORM_STATS_URL, TRANSFORM_STATES } from '../../../../common/constants';
+import { TransformStatsResponse } from './types';
 
 type EndpointMetadataHttpMocksInterface = ResponseProvidersInterface<{
   metadataList: () => HostResultList;
@@ -238,14 +238,14 @@ export const failedTransformStateMock = {
   count: 1,
   transforms: [
     {
-      state: TRANSFORM_STATE.FAILED,
+      state: TRANSFORM_STATES.FAILED,
     },
   ],
 };
 export const transformsHttpMocks = httpHandlerMockFactory<TransformHttpMocksInterface>([
   {
     id: 'metadataTransformStats',
-    path: TRANSFORM_STATS_URL,
+    path: METADATA_TRANSFORM_STATS_URL,
     method: 'get',
     handler: () => failedTransformStateMock,
   },

x-pack/plugins/security_solution/public/management/pages/endpoint_hosts/store/middleware.ts

@@ -78,7 +78,7 @@ import { resolvePathVariables } from '../../../../common/utils/resolve_path_vari
 import { EndpointPackageInfoStateChanged } from './action';
 import { fetchPendingActionsByAgentId } from '../../../../common/lib/endpoint_pending_actions';
 import { getIsInvalidDateRange } from '../utils';
-import { TRANSFORM_STATS_URL } from '../../../../../common/constants';
+import { METADATA_TRANSFORM_STATS_URL } from '../../../../../common/constants';
 
 type EndpointPageStore = ImmutableMiddlewareAPI<EndpointState, AppAction>;
 
@@ -785,7 +785,9 @@ export async function handleLoadMetadataTransformStats(http: HttpStart, store: E
   });
 
   try {
-    const transformStatsResponse: TransformStatsResponse = await http.get(TRANSFORM_STATS_URL);
+    const transformStatsResponse: TransformStatsResponse = await http.get(
+      METADATA_TRANSFORM_STATS_URL
+    );
 
     dispatch({
       type: 'metadataTransformStatsChanged',

x-pack/plugins/security_solution/public/management/pages/endpoint_hosts/store/mock_endpoint_result_list.ts

@@ -30,7 +30,7 @@ import {
 import { GetPolicyListResponse } from '../../policy/types';
 import { pendingActionsResponseMock } from '../../../../common/lib/endpoint_pending_actions/mocks';
 import { ACTION_STATUS_ROUTE } from '../../../../../common/endpoint/constants';
-import { TRANSFORM_STATS_URL } from '../../../../../common/constants';
+import { METADATA_TRANSFORM_STATS_URL } from '../../../../../common/constants';
 import { TransformStats, TransformStatsResponse } from '../types';
 
 const generator = new EndpointDocGenerator('seed');
@@ -163,7 +163,7 @@ const endpointListApiPathHandlerMocks = ({
       return pendingActionsResponseMock();
     },
 
-    [TRANSFORM_STATS_URL]: (): TransformStatsResponse => ({
+    [METADATA_TRANSFORM_STATS_URL]: (): TransformStatsResponse => ({
       count: transforms.length,
       transforms,
     }),

x-pack/plugins/security_solution/public/management/pages/endpoint_hosts/types.ts

@@ -22,6 +22,7 @@ import { ServerApiError } from '../../../common/types';
 import { GetPackagesResponse } from '../../../../../fleet/common';
 import { IIndexPattern } from '../../../../../../../src/plugins/data/public';
 import { AsyncResourceState } from '../../state';
+import { TRANSFORM_STATES } from '../../../../common/constants';
 
 export interface EndpointState {
   /** list of host **/
@@ -143,24 +144,7 @@ export interface EndpointIndexUIQueryParams {
   admin_query?: string;
 }
 
-export const TRANSFORM_STATE = {
-  ABORTING: 'aborting',
-  FAILED: 'failed',
-  INDEXING: 'indexing',
-  STARTED: 'started',
-  STOPPED: 'stopped',
-  STOPPING: 'stopping',
-  WAITING: 'waiting',
-};
-
-export const WARNING_TRANSFORM_STATES = new Set([
-  TRANSFORM_STATE.ABORTING,
-  TRANSFORM_STATE.FAILED,
-  TRANSFORM_STATE.STOPPED,
-  TRANSFORM_STATE.STOPPING,
-]);
-
-const transformStates = Object.values(TRANSFORM_STATE);
+const transformStates = Object.values(TRANSFORM_STATES);
 export type TransformState = typeof transformStates[number];
 
 export interface TransformStats {

x-pack/plugins/security_solution/public/management/pages/endpoint_hosts/view/index.test.tsx

@@ -46,8 +46,9 @@ import {
   APP_PATH,
   MANAGEMENT_PATH,
   DEFAULT_TIMEPICKER_QUICK_RANGES,
+  TRANSFORM_STATES,
 } from '../../../../../common/constants';
-import { TransformStats, TRANSFORM_STATE } from '../types';
+import { TransformStats } from '../types';
 import { metadataTransformPrefix } from '../../../../../common/endpoint/constants';
 
 // not sure why this can't be imported from '../../../../common/mock/formatted_relative';
@@ -1403,7 +1404,7 @@ describe('when on the endpoint list page', () => {
       const transforms: TransformStats[] = [
         {
           id: `${metadataTransformPrefix}-0.20.0`,
-          state: TRANSFORM_STATE.STARTED,
+          state: TRANSFORM_STATES.STARTED,
         } as TransformStats,
       ];
       setEndpointListApiMockImplementation(coreStart.http, { transforms });
@@ -1414,7 +1415,7 @@ describe('when on the endpoint list page', () => {
 
     it('is not displayed when non-relevant transform is failing', () => {
       const transforms: TransformStats[] = [
-        { id: 'not-metadata', state: TRANSFORM_STATE.FAILED } as TransformStats,
+        { id: 'not-metadata', state: TRANSFORM_STATES.FAILED } as TransformStats,
       ];
       setEndpointListApiMockImplementation(coreStart.http, { transforms });
       render();
@@ -1426,7 +1427,7 @@ describe('when on the endpoint list page', () => {
       const transforms: TransformStats[] = [
         {
           id: `${metadataTransformPrefix}-0.20.0`,
-          state: TRANSFORM_STATE.FAILED,
+          state: TRANSFORM_STATES.FAILED,
         } as TransformStats,
       ];
       setEndpointListApiMockImplementation(coreStart.http, { transforms });

x-pack/plugins/security_solution/public/management/pages/endpoint_hosts/view/index.tsx

@@ -58,8 +58,8 @@ import { LinkToApp } from '../../../../common/components/endpoint/link_to_app';
 import { TableRowActions } from './components/table_row_actions';
 import { EndpointAgentStatus } from './components/endpoint_agent_status';
 import { CallOut } from '../../../../common/components/callouts';
-import { WARNING_TRANSFORM_STATES } from '../types';
 import { metadataTransformPrefix } from '../../../../../common/endpoint/constants';
+import { WARNING_TRANSFORM_STATES } from '../../../../../common/constants';
 
 const MAX_PAGINATED_ITEM = 9999;
 const TRANSFORM_URL = '/data/transform';

x-pack/plugins/security_solution/server/endpoint/lib/metadata/check_metadata_transforms_task.test.ts

@@ -0,0 +1,250 @@
+/*
+ * 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 { ApiResponse } from '@elastic/elasticsearch';
+import { TransformGetTransformStatsResponse } from '@elastic/elasticsearch/api/types';
+import {
+  CheckMetadataTransformsTask,
+  TYPE,
+  VERSION,
+  BASE_NEXT_ATTEMPT_DELAY,
+} from './check_metadata_transforms_task';
+import { createMockEndpointAppContext } from '../../mocks';
+import { coreMock } from '../../../../../../../src/core/server/mocks';
+import { taskManagerMock } from '../../../../../task_manager/server/mocks';
+import { TaskManagerSetupContract, TaskStatus } from '../../../../../task_manager/server';
+import { CoreSetup } from '../../../../../../../src/core/server';
+// eslint-disable-next-line @kbn/eslint/no-restricted-paths
+import { ElasticsearchClientMock } from '../../../../../../../src/core/server/elasticsearch/client/mocks';
+import { TRANSFORM_STATES } from '../../../../common/constants';
+import { METADATA_TRANSFORMS_PATTERN } from '../../../../common/endpoint/constants';
+import { RunResult } from '../../../../../task_manager/server/task';
+
+const MOCK_TASK_INSTANCE = {
+  id: `${TYPE}:${VERSION}`,
+  runAt: new Date(),
+  attempts: 0,
+  ownerId: '',
+  status: TaskStatus.Running,
+  startedAt: new Date(),
+  scheduledAt: new Date(),
+  retryAt: new Date(),
+  params: {},
+  state: {},
+  taskType: TYPE,
+};
+const failedTransformId = 'failing-transform';
+const goodTransformId = 'good-transform';
+
+describe('check metadata transforms task', () => {
+  const { createSetup: coreSetupMock } = coreMock;
+  const { createSetup: tmSetupMock, createStart: tmStartMock } = taskManagerMock;
+
+  let mockTask: CheckMetadataTransformsTask;
+  let mockCore: CoreSetup;
+  let mockTaskManagerSetup: jest.Mocked<TaskManagerSetupContract>;
+  beforeAll(() => {
+    mockCore = coreSetupMock();
+    mockTaskManagerSetup = tmSetupMock();
+    mockTask = new CheckMetadataTransformsTask({
+      endpointAppContext: createMockEndpointAppContext(),
+      core: mockCore,
+      taskManager: mockTaskManagerSetup,
+    });
+  });
+
+  describe('task lifecycle', () => {
+    it('should create task', () => {
+      expect(mockTask).toBeInstanceOf(CheckMetadataTransformsTask);
+    });
+
+    it('should register task', () => {
+      expect(mockTaskManagerSetup.registerTaskDefinitions).toHaveBeenCalled();
+    });
+
+    it('should schedule task', async () => {
+      const mockTaskManagerStart = tmStartMock();
+      await mockTask.start({ taskManager: mockTaskManagerStart });
+      expect(mockTaskManagerStart.ensureScheduled).toHaveBeenCalled();
+    });
+  });
+
+  describe('task logic', () => {
+    let esClient: ElasticsearchClientMock;
+    beforeEach(async () => {
+      const [{ elasticsearch }] = await mockCore.getStartServices();
+      esClient = elasticsearch.client.asInternalUser as ElasticsearchClientMock;
+    });
+
+    const runTask = async (taskInstance = MOCK_TASK_INSTANCE) => {
+      const mockTaskManagerStart = tmStartMock();
+      await mockTask.start({ taskManager: mockTaskManagerStart });
+      const createTaskRunner =
+        mockTaskManagerSetup.registerTaskDefinitions.mock.calls[0][0][TYPE].createTaskRunner;
+      const taskRunner = createTaskRunner({ taskInstance });
+      return taskRunner.run();
+    };
+
+    const buildFailedStatsResponse = () =>
+      ({
+        body: {
+          transforms: [
+            {
+              id: goodTransformId,
+              state: TRANSFORM_STATES.STARTED,
+            },
+            {
+              id: failedTransformId,
+              state: TRANSFORM_STATES.FAILED,
+            },
+          ],
+        },
+      } as unknown as ApiResponse<TransformGetTransformStatsResponse>);
+
+    it('should stop task if transform stats response fails', async () => {
+      esClient.transform.getTransformStats.mockRejectedValue({});
+      await runTask();
+      expect(esClient.transform.getTransformStats).toHaveBeenCalledWith({
+        transform_id: METADATA_TRANSFORMS_PATTERN,
+      });
+      expect(esClient.transform.stopTransform).not.toHaveBeenCalled();
+      expect(esClient.transform.startTransform).not.toHaveBeenCalled();
+    });
+
+    it('should attempt transform restart if failing state', async () => {
+      const transformStatsResponseMock = buildFailedStatsResponse();
+      esClient.transform.getTransformStats.mockResolvedValue(transformStatsResponseMock);
+
+      const taskResponse = (await runTask()) as RunResult;
+
+      expect(esClient.transform.getTransformStats).toHaveBeenCalledWith({
+        transform_id: METADATA_TRANSFORMS_PATTERN,
+      });
+      expect(esClient.transform.stopTransform).toHaveBeenCalledWith({
+        transform_id: failedTransformId,
+        allow_no_match: true,
+        wait_for_completion: true,
+        force: true,
+      });
+      expect(esClient.transform.startTransform).toHaveBeenCalledWith({
+        transform_id: failedTransformId,
+      });
+      expect(taskResponse?.state?.attempts).toEqual({
+        [goodTransformId]: 0,
+        [failedTransformId]: 0,
+      });
+    });
+
+    it('should correctly track transform restart attempts', async () => {
+      const transformStatsResponseMock = buildFailedStatsResponse();
+      esClient.transform.getTransformStats.mockResolvedValue(transformStatsResponseMock);
+
+      esClient.transform.stopTransform.mockRejectedValueOnce({});
+      let taskResponse = (await runTask()) as RunResult;
+      expect(taskResponse?.state?.attempts).toEqual({
+        [goodTransformId]: 0,
+        [failedTransformId]: 1,
+      });
+
+      esClient.transform.startTransform.mockRejectedValueOnce({});
+      taskResponse = (await runTask({
+        ...MOCK_TASK_INSTANCE,
+        state: taskResponse.state,
+      })) as RunResult;
+      expect(taskResponse?.state?.attempts).toEqual({
+        [goodTransformId]: 0,
+        [failedTransformId]: 2,
+      });
+
+      taskResponse = (await runTask({
+        ...MOCK_TASK_INSTANCE,
+        state: taskResponse.state,
+      })) as RunResult;
+      expect(taskResponse?.state?.attempts).toEqual({
+        [goodTransformId]: 0,
+        [failedTransformId]: 0,
+      });
+    });
+
+    it('should correctly back off subsequent restart attempts', async () => {
+      let transformStatsResponseMock = buildFailedStatsResponse();
+      esClient.transform.getTransformStats.mockResolvedValue(transformStatsResponseMock);
+
+      esClient.transform.stopTransform.mockRejectedValueOnce({});
+      let taskStartedAt = new Date();
+      let taskResponse = (await runTask()) as RunResult;
+      let delay = BASE_NEXT_ATTEMPT_DELAY * 60000;
+      let expectedRunAt = taskStartedAt.getTime() + delay;
+      expect(taskResponse?.runAt?.getTime()).toBeGreaterThanOrEqual(expectedRunAt);
+      // we don't have the exact timestamp it uses so give a buffer
+      let expectedRunAtUpperBound = expectedRunAt + 1000;
+      expect(taskResponse?.runAt?.getTime()).toBeLessThanOrEqual(expectedRunAtUpperBound);
+
+      esClient.transform.startTransform.mockRejectedValueOnce({});
+      taskStartedAt = new Date();
+      taskResponse = (await runTask({
+        ...MOCK_TASK_INSTANCE,
+        state: taskResponse.state,
+      })) as RunResult;
+      // should be exponential on second+ attempt
+      delay = BASE_NEXT_ATTEMPT_DELAY ** 2 * 60000;
+      expectedRunAt = taskStartedAt.getTime() + delay;
+      expect(taskResponse?.runAt?.getTime()).toBeGreaterThanOrEqual(expectedRunAt);
+      // we don't have the exact timestamp it uses so give a buffer
+      expectedRunAtUpperBound = expectedRunAt + 1000;
+      expect(taskResponse?.runAt?.getTime()).toBeLessThanOrEqual(expectedRunAtUpperBound);
+
+      esClient.transform.stopTransform.mockRejectedValueOnce({});
+      taskStartedAt = new Date();
+      taskResponse = (await runTask({
+        ...MOCK_TASK_INSTANCE,
+        state: taskResponse.state,
+      })) as RunResult;
+      // should be exponential on second+ attempt
+      delay = BASE_NEXT_ATTEMPT_DELAY ** 3 * 60000;
+      expectedRunAt = taskStartedAt.getTime() + delay;
+      expect(taskResponse?.runAt?.getTime()).toBeGreaterThanOrEqual(expectedRunAt);
+      // we don't have the exact timestamp it uses so give a buffer
+      expectedRunAtUpperBound = expectedRunAt + 1000;
+      expect(taskResponse?.runAt?.getTime()).toBeLessThanOrEqual(expectedRunAtUpperBound);
+
+      taskStartedAt = new Date();
+      taskResponse = (await runTask({
+        ...MOCK_TASK_INSTANCE,
+        state: taskResponse.state,
+      })) as RunResult;
+      // back to base delay after success
+      delay = BASE_NEXT_ATTEMPT_DELAY * 60000;
+      expectedRunAt = taskStartedAt.getTime() + delay;
+      expect(taskResponse?.runAt?.getTime()).toBeGreaterThanOrEqual(expectedRunAt);
+      // we don't have the exact timestamp it uses so give a buffer
+      expectedRunAtUpperBound = expectedRunAt + 1000;
+      expect(taskResponse?.runAt?.getTime()).toBeLessThanOrEqual(expectedRunAtUpperBound);
+
+      transformStatsResponseMock = {
+        body: {
+          transforms: [
+            {
+              id: goodTransformId,
+              state: TRANSFORM_STATES.STARTED,
+            },
+            {
+              id: failedTransformId,
+              state: TRANSFORM_STATES.STARTED,
+            },
+          ],
+        },
+      } as unknown as ApiResponse<TransformGetTransformStatsResponse>;
+      esClient.transform.getTransformStats.mockResolvedValue(transformStatsResponseMock);
+      taskResponse = (await runTask({
+        ...MOCK_TASK_INSTANCE,
+        state: taskResponse.state,
+      })) as RunResult;
+      // no more explicit runAt after subsequent success
+      expect(taskResponse?.runAt).toBeUndefined();
+    });
+  });
+});

x-pack/plugins/security_solution/server/endpoint/lib/metadata/check_metadata_transforms_task.ts

@@ -0,0 +1,214 @@
+/*
+ * 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 { ApiResponse } from '@elastic/elasticsearch';
+import {
+  TransformGetTransformStatsResponse,
+  TransformGetTransformStatsTransformStats,
+} from '@elastic/elasticsearch/api/types';
+import { CoreSetup, ElasticsearchClient, Logger } from 'src/core/server';
+import {
+  ConcreteTaskInstance,
+  TaskManagerSetupContract,
+  TaskManagerStartContract,
+  throwUnrecoverableError,
+} from '../../../../../task_manager/server';
+import { EndpointAppContext } from '../../types';
+import { METADATA_TRANSFORMS_PATTERN } from '../../../../common/endpoint/constants';
+import { WARNING_TRANSFORM_STATES } from '../../../../common/constants';
+import { wrapErrorIfNeeded } from '../../utils';
+
+const SCOPE = ['securitySolution'];
+const INTERVAL = '2h';
+const TIMEOUT = '4m';
+export const TYPE = 'endpoint:metadata-check-transforms-task';
+export const VERSION = '0.0.1';
+const MAX_ATTEMPTS = 5;
+export const BASE_NEXT_ATTEMPT_DELAY = 5; // minutes
+
+export interface CheckMetadataTransformsTaskSetupContract {
+  endpointAppContext: EndpointAppContext;
+  core: CoreSetup;
+  taskManager: TaskManagerSetupContract;
+}
+
+export interface CheckMetadataTransformsTaskStartContract {
+  taskManager: TaskManagerStartContract;
+}
+
+export class CheckMetadataTransformsTask {
+  private logger: Logger;
+  private wasStarted: boolean = false;
+
+  constructor(setupContract: CheckMetadataTransformsTaskSetupContract) {
+    const { endpointAppContext, core, taskManager } = setupContract;
+    this.logger = endpointAppContext.logFactory.get(this.getTaskId());
+    taskManager.registerTaskDefinitions({
+      [TYPE]: {
+        title: 'Security Solution Endpoint Metadata Periodic Tasks',
+        timeout: TIMEOUT,
+        createTaskRunner: ({ taskInstance }: { taskInstance: ConcreteTaskInstance }) => {
+          return {
+            run: async () => {
+              return this.runTask(taskInstance, core);
+            },
+            cancel: async () => {},
+          };
+        },
+      },
+    });
+  }
+
+  public start = async ({ taskManager }: CheckMetadataTransformsTaskStartContract) => {
+    if (!taskManager) {
+      this.logger.error('missing required service during start');
+      return;
+    }
+
+    this.wasStarted = true;
+
+    try {
+      await taskManager.ensureScheduled({
+        id: this.getTaskId(),
+        taskType: TYPE,
+        scope: SCOPE,
+        schedule: {
+          interval: INTERVAL,
+        },
+        state: {
+          attempts: {},
+        },
+        params: { version: VERSION },
+      });
+    } catch (e) {
+      this.logger.debug(`Error scheduling task, received ${e.message}`);
+    }
+  };
+
+  private runTask = async (taskInstance: ConcreteTaskInstance, core: CoreSetup) => {
+    // if task was not `.start()`'d yet, then exit
+    if (!this.wasStarted) {
+      this.logger.debug('[runTask()] Aborted. MetadataTask not started yet');
+      return;
+    }
+
+    // Check that this task is current
+    if (taskInstance.id !== this.getTaskId()) {
+      // old task, die
+      throwUnrecoverableError(new Error('Outdated task version'));
+    }
+
+    const [{ elasticsearch }] = await core.getStartServices();
+    const esClient = elasticsearch.client.asInternalUser;
+
+    let transformStatsResponse: ApiResponse<TransformGetTransformStatsResponse>;
+    try {
+      transformStatsResponse = await esClient?.transform.getTransformStats({
+        transform_id: METADATA_TRANSFORMS_PATTERN,
+      });
+    } catch (e) {
+      const err = wrapErrorIfNeeded(e);
+      const errMessage = `failed to get transform stats with error: ${err}`;
+      this.logger.error(errMessage);
+
+      return;
+    }
+
+    const { transforms } = transformStatsResponse.body;
+    if (!transforms.length) {
+      this.logger.info('no OLM metadata transforms found');
+      return;
+    }
+
+    let didAttemptRestart: boolean = false;
+    let highestAttempt: number = 0;
+    const attempts = { ...taskInstance.state.attempts };
+
+    for (const transform of transforms) {
+      const restartedTransform = await this.restartTransform(
+        esClient,
+        transform,
+        attempts[transform.id]
+      );
+      if (restartedTransform.didAttemptRestart) {
+        didAttemptRestart = true;
+      }
+      attempts[transform.id] = restartedTransform.attempts;
+      highestAttempt = Math.max(attempts[transform.id], highestAttempt);
+    }
+
+    // after a restart attempt run next check sooner with exponential backoff
+    let runAt: Date | undefined;
+    if (didAttemptRestart) {
+      const delay = BASE_NEXT_ATTEMPT_DELAY ** Math.max(highestAttempt, 1) * 60000;
+      runAt = new Date(new Date().getTime() + delay);
+    }
+
+    const nextState = { attempts };
+    const nextTask = runAt ? { state: nextState, runAt } : { state: nextState };
+    return nextTask;
+  };
+
+  private restartTransform = async (
+    esClient: ElasticsearchClient,
+    transform: TransformGetTransformStatsTransformStats,
+    currentAttempts: number = 0
+  ) => {
+    let attempts = currentAttempts;
+    let didAttemptRestart = false;
+
+    if (!WARNING_TRANSFORM_STATES.has(transform.state)) {
+      return {
+        attempts,
+        didAttemptRestart,
+      };
+    }
+
+    if (attempts > MAX_ATTEMPTS) {
+      this.logger.warn(
+        `transform ${transform.id} has failed to restart ${attempts} times. stopping auto restart attempts.`
+      );
+      return {
+        attempts,
+        didAttemptRestart,
+      };
+    }
+
+    try {
+      this.logger.info(`failed transform detected with id: ${transform.id}. attempting restart.`);
+      await esClient.transform.stopTransform({
+        transform_id: transform.id,
+        allow_no_match: true,
+        wait_for_completion: true,
+        force: true,
+      });
+      await esClient.transform.startTransform({
+        transform_id: transform.id,
+      });
+
+      // restart succeeded, reset attempt count
+      attempts = 0;
+    } catch (e) {
+      const err = wrapErrorIfNeeded(e);
+      const errMessage = `failed to restart transform ${transform.id} with error: ${err}`;
+      this.logger.error(errMessage);
+
+      // restart failed, increment attempt count
+      attempts = attempts + 1;
+    } finally {
+      didAttemptRestart = true;
+    }
+
+    return {
+      attempts,
+      didAttemptRestart,
+    };
+  };
+
+  private getTaskId = (): string => {
+    return `${TYPE}:${VERSION}`;
+  };
+}

x-pack/plugins/security_solution/server/endpoint/lib/metadata/index.ts

@@ -0,0 +1,8 @@
+/*
+ * 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.
+ */
+
+export * from './check_metadata_transforms_task';

x-pack/plugins/security_solution/server/plugin.ts

@@ -59,6 +59,7 @@ import { initRoutes } from './routes';
 import { isAlertExecutor } from './lib/detection_engine/signals/types';
 import { signalRulesAlertType } from './lib/detection_engine/signals/signal_rule_alert_type';
 import { ManifestTask } from './endpoint/lib/artifacts';
+import { CheckMetadataTransformsTask } from './endpoint/lib/metadata';
 import { initSavedObjects } from './saved_objects';
 import { AppClientFactory } from './client';
 import { createConfig, ConfigType } from './config';
@@ -157,6 +158,7 @@ export class Plugin implements IPlugin<PluginSetup, PluginStart, SetupPlugins, S
   private policyWatcher?: PolicyWatcher;
 
   private manifestTask: ManifestTask | undefined;
+  private checkMetadataTransformsTask: CheckMetadataTransformsTask | undefined;
   private artifactsCache: LRU<string, Buffer>;
   private telemetryUsageCounter?: UsageCounter;
 
@@ -362,6 +364,12 @@ export class Plugin implements IPlugin<PluginSetup, PluginStart, SetupPlugins, S
       this.telemetryUsageCounter
     );
 
+    this.checkMetadataTransformsTask = new CheckMetadataTransformsTask({
+      endpointAppContext: endpointContext,
+      core,
+      taskManager: plugins.taskManager!,
+    });
+
     return {};
   }
 
@@ -451,6 +459,10 @@ export class Plugin implements IPlugin<PluginSetup, PluginStart, SetupPlugins, S
       this.telemetryReceiver
     );
 
+    this.checkMetadataTransformsTask?.start({
+      taskManager: plugins.taskManager!,
+    });
+
     return {};
   }