Cleanup action task params objects after successful execution

x-pack/legacy/plugins/actions/mappings.json

@@ -27,6 +27,9 @@
       },
       "apiKey": {
         "type": "binary"
+      },
+      "createdAt": {
+        "type": "date"
       }
     }
   }

x-pack/legacy/plugins/actions/server/create_execute_function.test.ts

@@ -4,14 +4,21 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
+import sinon from 'sinon';
 import { taskManagerMock } from '../../../../plugins/task_manager/server/task_manager.mock';
 import { createExecuteFunction } from './create_execute_function';
 import { savedObjectsClientMock } from '../../../../../src/core/server/mocks';
 
+let fakeTimer: sinon.SinonFakeTimers;
 const mockTaskManager = taskManagerMock.start();
 const savedObjectsClient = savedObjectsClientMock.create();
 const getBasePath = jest.fn();
 
+beforeAll(() => {
+  fakeTimer = sinon.useFakeTimers();
+});
+afterAll(() => fakeTimer.restore());
+
 beforeEach(() => jest.resetAllMocks());
 
 describe('execute()', () => {
@@ -62,6 +69,7 @@ describe('execute()', () => {
       actionId: '123',
       params: { baz: false },
       apiKey: Buffer.from('123:abc').toString('base64'),
+      createdAt: '1970-01-01T00:00:00.000Z',
     });
   });
 

x-pack/legacy/plugins/actions/server/create_execute_function.ts

@@ -56,6 +56,7 @@ export function createExecuteFunction({
       actionId: id,
       params,
       apiKey,
+      createdAt: new Date().toISOString(),
     });
 
     await taskManager.schedule({

x-pack/legacy/plugins/actions/server/lib/action_executor.mock.ts

@@ -9,7 +9,7 @@ import { ActionExecutorContract } from './action_executor';
 const createActionExecutorMock = () => {
   const mocked: jest.Mocked<ActionExecutorContract> = {
     initialize: jest.fn(),
-    execute: jest.fn(),
+    execute: jest.fn().mockResolvedValue({ status: 'ok', actionId: '' }),
   };
   return mocked;
 };

x-pack/legacy/plugins/actions/server/lib/task_runner_factory.test.ts

@@ -65,13 +65,18 @@ const actionExecutorInitializerParams = {
 };
 const taskRunnerFactoryInitializerParams = {
   spaceIdToNamespace,
+  logger: loggingServiceMock.create().get(),
   encryptedSavedObjectsPlugin: mockedEncryptedSavedObjectsPlugin,
   getBasePath: jest.fn().mockReturnValue(undefined),
+  getScopedSavedObjectsClient: jest.fn().mockReturnValue(services.savedObjectsClient),
 };
 
 beforeEach(() => {
   jest.resetAllMocks();
   actionExecutorInitializerParams.getServices.mockReturnValue(services);
+  taskRunnerFactoryInitializerParams.getScopedSavedObjectsClient.mockReturnValue(
+    services.savedObjectsClient
+  );
 });
 
 test(`throws an error if factory isn't initialized`, () => {
@@ -137,6 +142,56 @@ test('executes the task by calling the executor with proper parameters', async (
   });
 });
 
+test('cleans up action_task_params object', async () => {
+  const taskRunner = taskRunnerFactory.create({
+    taskInstance: mockedTaskInstance,
+  });
+
+  mockedActionExecutor.execute.mockResolvedValueOnce({ status: 'ok', actionId: '2' });
+  spaceIdToNamespace.mockReturnValueOnce('namespace-test');
+  mockedEncryptedSavedObjectsPlugin.getDecryptedAsInternalUser.mockResolvedValueOnce({
+    id: '3',
+    type: 'action_task_params',
+    attributes: {
+      actionId: '2',
+      params: { baz: true },
+      apiKey: Buffer.from('123:abc').toString('base64'),
+    },
+    references: [],
+  });
+
+  await taskRunner.run();
+
+  expect(services.savedObjectsClient.delete).toHaveBeenCalledWith('action_task_params', '3');
+});
+
+test('runs successfully when cleanup fails and logs the error', async () => {
+  const taskRunner = taskRunnerFactory.create({
+    taskInstance: mockedTaskInstance,
+  });
+
+  mockedActionExecutor.execute.mockResolvedValueOnce({ status: 'ok', actionId: '2' });
+  spaceIdToNamespace.mockReturnValueOnce('namespace-test');
+  mockedEncryptedSavedObjectsPlugin.getDecryptedAsInternalUser.mockResolvedValueOnce({
+    id: '3',
+    type: 'action_task_params',
+    attributes: {
+      actionId: '2',
+      params: { baz: true },
+      apiKey: Buffer.from('123:abc').toString('base64'),
+    },
+    references: [],
+  });
+  services.savedObjectsClient.delete.mockRejectedValueOnce(new Error('Fail'));
+
+  await taskRunner.run();
+
+  expect(services.savedObjectsClient.delete).toHaveBeenCalledWith('action_task_params', '3');
+  expect(taskRunnerFactoryInitializerParams.logger.error).toHaveBeenCalledWith(
+    'Failed to cleanup action_task_params object [id="3"]: Fail'
+  );
+});
+
 test('throws an error with suggested retry logic when return status is error', async () => {
   const taskRunner = taskRunnerFactory.create({
     taskInstance: mockedTaskInstance,

x-pack/legacy/plugins/actions/server/lib/task_runner_factory.ts

@@ -6,14 +6,18 @@
 
 import { ActionExecutorContract } from './action_executor';
 import { ExecutorError } from './executor_error';
+import { ActionsCoreStart } from '../shim';
+import { Logger } from '../../../../../../src/core/server';
 import { RunContext } from '../../../../../plugins/task_manager/server';
 import { PluginStartContract as EncryptedSavedObjectsStartContract } from '../../../../../plugins/encrypted_saved_objects/server';
 import { ActionTaskParams, GetBasePathFunction, SpaceIdToNamespaceFunction } from '../types';
 
 export interface TaskRunnerContext {
+  logger: Logger;
   encryptedSavedObjectsPlugin: EncryptedSavedObjectsStartContract;
   spaceIdToNamespace: SpaceIdToNamespaceFunction;
   getBasePath: GetBasePathFunction;
+  getScopedSavedObjectsClient: ActionsCoreStart['savedObjects']['getScopedSavedObjectsClient'];
 }
 
 export class TaskRunnerFactory {
@@ -40,9 +44,11 @@ export class TaskRunnerFactory {
 
     const { actionExecutor } = this;
     const {
+      logger,
       encryptedSavedObjectsPlugin,
       spaceIdToNamespace,
       getBasePath,
+      getScopedSavedObjectsClient,
     } = this.taskRunnerContext!;
 
     return {
@@ -93,6 +99,17 @@ export class TaskRunnerFactory {
             executorResult.data,
             executorResult.retry == null ? false : executorResult.retry
           );
+        } else if (executorResult.status === 'ok') {
+          // Cleanup action_task_params object now that we're done with it
+          try {
+            const savedObjectsClient = getScopedSavedObjectsClient(fakeRequest);
+            await savedObjectsClient.delete('action_task_params', actionTaskParamsId);
+          } catch (e) {
+            // Log error only, we shouldn't fail the task because of an error here (if ever there's retry logic)
+            logger.error(
+              `Failed to cleanup action_task_params object [id="${actionTaskParamsId}"]: ${e.message}`
+            );
+          }
         }
       },
     };

x-pack/legacy/plugins/actions/server/plugin.ts

@@ -86,6 +86,7 @@ export class Plugin {
     plugins.encryptedSavedObjects.registerType({
       type: 'action_task_params',
       attributesToEncrypt: new Set(['apiKey']),
+      attributesToExcludeFromAAD: new Set(['createdAt']),
     });
 
     const actionExecutor = new ActionExecutor();
@@ -161,6 +162,7 @@ export class Plugin {
       encryptedSavedObjectsPlugin: plugins.encryptedSavedObjects,
       getBasePath,
       spaceIdToNamespace,
+      getScopedSavedObjectsClient: core.savedObjects.getScopedSavedObjectsClient,
     });
 
     const executeFn = createExecuteFunction({

x-pack/legacy/plugins/actions/server/types.ts

@@ -93,4 +93,5 @@ export interface ActionTaskParams extends SavedObjectAttributes {
   actionId: string;
   params: Record<string, any>;
   apiKey?: string;
+  createdAt: string;
 }

x-pack/test/alerting_api_integration/common/fixtures/plugins/alerts/index.ts

@@ -62,8 +62,8 @@ export default function(kibana: any) {
             encrypted: schema.string(),
           }),
         },
-        async executor({ config, secrets, params, services }: ActionTypeExecutorOptions) {
-          return await services.callCluster('index', {
+        async executor({ config, secrets, params, services, actionId }: ActionTypeExecutorOptions) {
+          await services.callCluster('index', {
             index: params.index,
             refresh: 'wait_for',
             body: {
@@ -74,6 +74,7 @@ export default function(kibana: any) {
               source: 'action:test.index-record',
             },
           });
+          return { status: 'ok', actionId };
         },
       };
       const failingActionType: ActionType = {
@@ -141,7 +142,7 @@ export default function(kibana: any) {
             reference: schema.string(),
           }),
         },
-        async executor({ params, services }: ActionTypeExecutorOptions) {
+        async executor({ params, services, actionId }: ActionTypeExecutorOptions) {
           // Call cluster
           let callClusterSuccess = false;
           let callClusterError;
@@ -186,8 +187,8 @@ export default function(kibana: any) {
             },
           });
           return {
+            actionId,
             status: 'ok',
-            actionId: '',
           };
         },
       };

x-pack/test/alerting_api_integration/common/lib/task_manager_utils.ts

@@ -43,4 +43,37 @@ export class TaskManagerUtils {
       }
     });
   }
+
+  async waitForActionTaskParamsToBeCleanedUp(createdAtFilter: Date): Promise<void> {
+    return await this.retry.try(async () => {
+      const searchResult = await this.es.search({
+        index: '.kibana',
+        body: {
+          query: {
+            bool: {
+              must: [
+                {
+                  term: {
+                    type: 'action_task_params',
+                  },
+                },
+                {
+                  range: {
+                    'action_task_params.createdAt': {
+                      gte: createdAtFilter,
+                    },
+                  },
+                },
+              ],
+            },
+          },
+        },
+      });
+      if (searchResult.hits.total.value) {
+        throw new Error(
+          `Expected 0 action_task_params objects but received ${searchResult.hits.total.value}`
+        );
+      }
+    });
+  }
 }

x-pack/test/alerting_api_integration/security_and_spaces/tests/alerting/alerts.ts

@@ -147,6 +147,8 @@ export default function alertTests({ getService }: FtrProviderContext) {
                 reference,
                 source: 'action:test.index-record',
               });
+
+              await taskManagerUtils.waitForActionTaskParamsToBeCleanedUp(testStart);
               break;
             default:
               throw new Error(`Scenario untested: ${JSON.stringify(scenario)}`);

x-pack/test/alerting_api_integration/spaces_only/tests/alerting/alerts_base.ts

@@ -16,6 +16,7 @@ import {
   ObjectRemover,
   AlertUtils,
   ensureDatetimeIsWithinRange,
+  TaskManagerUtils,
 } from '../../../common/lib';
 
 // eslint-disable-next-line import/no-default-export
@@ -24,6 +25,7 @@ export function alertTests({ getService }: FtrProviderContext, space: Space) {
   const es = getService('legacyEs');
   const retry = getService('retry');
   const esTestIndexTool = new ESTestIndexTool(es, retry);
+  const taskManagerUtils = new TaskManagerUtils(es, retry);
 
   function getAlertingTaskById(taskId: string) {
     return supertestWithoutAuth
@@ -73,6 +75,7 @@ export function alertTests({ getService }: FtrProviderContext, space: Space) {
     });
 
     it('should schedule task, run alert and schedule actions', async () => {
+      const testStart = new Date();
       const reference = alertUtils.generateReference();
       const response = await alertUtils.createAlwaysFiringAction({ reference });
       const alertId = response.body.id;
@@ -121,6 +124,8 @@ export function alertTests({ getService }: FtrProviderContext, space: Space) {
         reference,
         source: 'action:test.index-record',
       });
+
+      await taskManagerUtils.waitForActionTaskParamsToBeCleanedUp(testStart);
     });
 
     it('should reschedule failing alerts using the alerting interval and not the Task Manager retry logic', async () => {