[Alerting] Change execution of alerts from async to sync

docs/settings/alert-action-settings.asciidoc

@@ -207,4 +207,10 @@ Use `full` to perform hostname verification, `certificate` to skip hostname veri
 [[alert-settings]]
 ==== Alerting settings
 
-You do not need to configure any additional settings to use alerting in {kib}.
+[cols="2*<"]
+|===
+
+| `xpack.alerting.maxEphemeralActionsPerAlert`
+  | Sets the number of actions that will be executed ephemerally. To use this, enable ephemeral tasks in task manager first with <<task-manager-settings,`xpack.task_manager.ephemeral_tasks.enabled`>>
+
+|===

docs/settings/task-manager-settings.asciidoc

@@ -37,6 +37,14 @@ Task Manager runs background tasks by polling for work on an interval.  You can
   `monitored_stats_health_verbose_log.`
   `warn_delayed_task_start_in_seconds`
   | The amount of seconds we allow a task to delay before printing a warning server log.  Defaults to 60.
+
+  | `xpack.task_manager.ephemeral_tasks.enabled`
+  | Enables an experimental feature that executes a limited (and configurable) number of actions in the same task as the alert which triggered them.
+    These action tasks will reduce the latency of the time it takes an action to run after it's triggered, but are not persisted as SavedObjects.
+    These non-persisted action tasks have a risk that they won't be run at all if the Kibana instance running them exits unexpectedly. Defaults to false.
+
+  | `xpack.task_manager.ephemeral_tasks.request_capacity`
+  | Sets the size of the ephemeral queue defined above. Defaults to 10.
 |===
 
 [float]

x-pack/plugins/actions/server/actions_client.mock.ts

@@ -21,6 +21,7 @@ const createActionsClientMock = () => {
     getBulk: jest.fn(),
     execute: jest.fn(),
     enqueueExecution: jest.fn(),
+    ephemeralEnqueuedExecution: jest.fn(),
     listTypes: jest.fn(),
     isActionTypeEnabled: jest.fn(),
   };

x-pack/plugins/actions/server/actions_client.test.ts

@@ -44,6 +44,7 @@ const scopedClusterClient = elasticsearchServiceMock.createScopedClusterClient()
 const actionExecutor = actionExecutorMock.create();
 const authorization = actionsAuthorizationMock.create();
 const executionEnqueuer = jest.fn();
+const ephemeralExecutionEnqueuer = jest.fn();
 const request = httpServerMock.createKibanaRequest();
 const auditLogger = auditServiceMock.create().asScoped(request);
 
@@ -77,6 +78,7 @@ beforeEach(() => {
     preconfiguredActions: [],
     actionExecutor,
     executionEnqueuer,
+    ephemeralExecutionEnqueuer,
     request,
     authorization: (authorization as unknown) as ActionsAuthorization,
     auditLogger,
@@ -453,6 +455,7 @@ describe('create()', () => {
       preconfiguredActions: [],
       actionExecutor,
       executionEnqueuer,
+      ephemeralExecutionEnqueuer,
       request,
       authorization: (authorization as unknown) as ActionsAuthorization,
     });
@@ -553,6 +556,7 @@ describe('get()', () => {
         defaultKibanaIndex,
         actionExecutor,
         executionEnqueuer,
+        ephemeralExecutionEnqueuer,
         request,
         authorization: (authorization as unknown) as ActionsAuthorization,
         preconfiguredActions: [
@@ -608,6 +612,7 @@ describe('get()', () => {
         defaultKibanaIndex,
         actionExecutor,
         executionEnqueuer,
+        ephemeralExecutionEnqueuer,
         request,
         authorization: (authorization as unknown) as ActionsAuthorization,
         preconfiguredActions: [
@@ -724,6 +729,7 @@ describe('get()', () => {
       defaultKibanaIndex,
       actionExecutor,
       executionEnqueuer,
+      ephemeralExecutionEnqueuer,
       request,
       authorization: (authorization as unknown) as ActionsAuthorization,
       preconfiguredActions: [
@@ -793,6 +799,7 @@ describe('getAll()', () => {
         defaultKibanaIndex,
         actionExecutor,
         executionEnqueuer,
+        ephemeralExecutionEnqueuer,
         request,
         authorization: (authorization as unknown) as ActionsAuthorization,
         preconfiguredActions: [
@@ -930,6 +937,7 @@ describe('getAll()', () => {
       defaultKibanaIndex,
       actionExecutor,
       executionEnqueuer,
+      ephemeralExecutionEnqueuer,
       request,
       authorization: (authorization as unknown) as ActionsAuthorization,
       preconfiguredActions: [
@@ -1005,6 +1013,7 @@ describe('getBulk()', () => {
         defaultKibanaIndex,
         actionExecutor,
         executionEnqueuer,
+        ephemeralExecutionEnqueuer,
         request,
         authorization: (authorization as unknown) as ActionsAuthorization,
         preconfiguredActions: [
@@ -1136,6 +1145,7 @@ describe('getBulk()', () => {
       defaultKibanaIndex,
       actionExecutor,
       executionEnqueuer,
+      ephemeralExecutionEnqueuer,
       request,
       authorization: (authorization as unknown) as ActionsAuthorization,
       preconfiguredActions: [

x-pack/plugins/actions/server/actions_client.ts

@@ -41,6 +41,7 @@ import {
   AuthorizationMode,
 } from './authorization/get_authorization_mode_by_source';
 import { connectorAuditEvent, ConnectorAuditAction } from './lib/audit_events';
+import { RunNowResult } from '../../task_manager/server';
 
 // We are assuming there won't be many actions. This is why we will load
 // all the actions in advance and assume the total count to not go over 10000.
@@ -68,7 +69,8 @@ interface ConstructorOptions {
   unsecuredSavedObjectsClient: SavedObjectsClientContract;
   preconfiguredActions: PreConfiguredAction[];
   actionExecutor: ActionExecutorContract;
-  executionEnqueuer: ExecutionEnqueuer;
+  executionEnqueuer: ExecutionEnqueuer<void>;
+  ephemeralExecutionEnqueuer: ExecutionEnqueuer<RunNowResult>;
   request: KibanaRequest;
   authorization: ActionsAuthorization;
   auditLogger?: AuditLogger;
@@ -88,7 +90,8 @@ export class ActionsClient {
   private readonly actionExecutor: ActionExecutorContract;
   private readonly request: KibanaRequest;
   private readonly authorization: ActionsAuthorization;
-  private readonly executionEnqueuer: ExecutionEnqueuer;
+  private readonly executionEnqueuer: ExecutionEnqueuer<void>;
+  private readonly ephemeralExecutionEnqueuer: ExecutionEnqueuer<RunNowResult>;
   private readonly auditLogger?: AuditLogger;
 
   constructor({
@@ -99,6 +102,7 @@ export class ActionsClient {
     preconfiguredActions,
     actionExecutor,
     executionEnqueuer,
+    ephemeralExecutionEnqueuer,
     request,
     authorization,
     auditLogger,
@@ -110,6 +114,7 @@ export class ActionsClient {
     this.preconfiguredActions = preconfiguredActions;
     this.actionExecutor = actionExecutor;
     this.executionEnqueuer = executionEnqueuer;
+    this.ephemeralExecutionEnqueuer = ephemeralExecutionEnqueuer;
     this.request = request;
     this.authorization = authorization;
     this.auditLogger = auditLogger;
@@ -497,6 +502,17 @@ export class ActionsClient {
     return this.executionEnqueuer(this.unsecuredSavedObjectsClient, options);
   }
 
+  public async ephemeralEnqueuedExecution(options: EnqueueExecutionOptions): Promise<RunNowResult> {
+    const { source } = options;
+    if (
+      (await getAuthorizationModeBySource(this.unsecuredSavedObjectsClient, source)) ===
+      AuthorizationMode.RBAC
+    ) {
+      await this.authorization.ensureAuthorized('execute');
+    }
+    return this.ephemeralExecutionEnqueuer(this.unsecuredSavedObjectsClient, options);
+  }
+
   public async listTypes(): Promise<ActionType[]> {
     return this.actionTypeRegistry.list();
   }

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

@@ -6,8 +6,13 @@
  */
 
 import { SavedObjectsClientContract } from '../../../../src/core/server';
-import { TaskManagerStartContract } from '../../task_manager/server';
-import { RawAction, ActionTypeRegistryContract, PreConfiguredAction } from './types';
+import { RunNowResult, TaskManagerStartContract } from '../../task_manager/server';
+import {
+  RawAction,
+  ActionTypeRegistryContract,
+  PreConfiguredAction,
+  ActionTaskExecutorParams,
+} from './types';
 import { ACTION_TASK_PARAMS_SAVED_OBJECT_TYPE } from './constants/saved_objects';
 import { ExecuteOptions as ActionExecutorOptions } from './lib/action_executor';
 import { isSavedObjectExecutionSource } from './lib';
@@ -27,17 +32,17 @@ export interface ExecuteOptions extends Pick<ActionExecutorOptions, 'params' | '
   relatedSavedObjects?: RelatedSavedObjects;
 }
 
-export type ExecutionEnqueuer = (
+export type ExecutionEnqueuer<T> = (
   unsecuredSavedObjectsClient: SavedObjectsClientContract,
   options: ExecuteOptions
-) => Promise<void>;
+) => Promise<T>;
 
 export function createExecutionEnqueuerFunction({
   taskManager,
   actionTypeRegistry,
   isESOCanEncrypt,
   preconfiguredActions,
-}: CreateExecuteFunctionOptions) {
+}: CreateExecuteFunctionOptions): ExecutionEnqueuer<void> {
   return async function execute(
     unsecuredSavedObjectsClient: SavedObjectsClientContract,
     { id, params, spaceId, source, apiKey, relatedSavedObjects }: ExecuteOptions
@@ -48,18 +53,10 @@ export function createExecutionEnqueuerFunction({
       );
     }
 
-    const { actionTypeId, name, isMissingSecrets } = await getAction(
-      unsecuredSavedObjectsClient,
-      preconfiguredActions,
-      id
-    );
-
-    if (isMissingSecrets) {
-      throw new Error(
-        `Unable to execute action because no secrets are defined for the "${name}" connector.`
-      );
-    }
+    const action = await getAction(unsecuredSavedObjectsClient, preconfiguredActions, id);
+    validateCanActionBeUsed(action);
 
+    const { actionTypeId } = action;
     if (!actionTypeRegistry.isActionExecutable(id, actionTypeId, { notifyUsage: true })) {
       actionTypeRegistry.ensureActionTypeEnabled(actionTypeId);
     }
@@ -76,7 +73,7 @@ export function createExecutionEnqueuerFunction({
     );
 
     await taskManager.schedule({
-      taskType: `actions:${actionTypeId}`,
+      taskType: `actions:${action.actionTypeId}`,
       params: {
         spaceId,
         actionTaskParamsId: actionTaskParamsRecord.id,
@@ -87,6 +84,53 @@ export function createExecutionEnqueuerFunction({
   };
 }
 
+export function createEphemeralExecutionEnqueuerFunction({
+  taskManager,
+  actionTypeRegistry,
+  preconfiguredActions,
+}: CreateExecuteFunctionOptions): ExecutionEnqueuer<RunNowResult> {
+  return async function execute(
+    unsecuredSavedObjectsClient: SavedObjectsClientContract,
+    { id, params, spaceId, source, apiKey }: ExecuteOptions
+  ): Promise<RunNowResult> {
+    const action = await getAction(unsecuredSavedObjectsClient, preconfiguredActions, id);
+    validateCanActionBeUsed(action);
+
+    const { actionTypeId } = action;
+    if (!actionTypeRegistry.isActionExecutable(id, actionTypeId, { notifyUsage: true })) {
+      actionTypeRegistry.ensureActionTypeEnabled(actionTypeId);
+    }
+
+    const taskParams: ActionTaskExecutorParams = {
+      spaceId,
+      taskParams: {
+        actionId: id,
+        // Saved Objects won't allow us to enforce unknown rather than any
+        // eslint-disable-next-line @typescript-eslint/no-explicit-any
+        params: params as Record<string, any>,
+        ...(apiKey ? { apiKey } : {}),
+      },
+      ...executionSourceAsSavedObjectReferences(source),
+    };
+
+    return taskManager.ephemeralRunNow({
+      taskType: `actions:${action.actionTypeId}`,
+      params: taskParams,
+      state: {},
+      scope: ['actions'],
+    });
+  };
+}
+
+function validateCanActionBeUsed(action: PreConfiguredAction | RawAction) {
+  const { name, isMissingSecrets } = action;
+  if (isMissingSecrets) {
+    throw new Error(
+      `Unable to execute action because no secrets are defined for the "${name}" connector.`
+    );
+  }
+}
+
 function executionSourceAsSavedObjectReferences(executionSource: ActionExecutorOptions['source']) {
   return isSavedObjectExecutionSource(executionSource)
     ? {

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

@@ -48,6 +48,7 @@ export interface TaskInfo {
 
 export interface ExecuteOptions<Source = unknown> {
   actionId: string;
+  isEphemeral?: boolean;
   request: KibanaRequest;
   params: Record<string, unknown>;
   source?: ActionExecutionSource<Source>;
@@ -79,6 +80,7 @@ export class ActionExecutor {
     params,
     request,
     source,
+    isEphemeral,
     taskInfo,
     relatedSavedObjects,
   }: ExecuteOptions): Promise<ActionTypeExecutorResult<unknown>> {
@@ -207,6 +209,7 @@ export class ActionExecutor {
             params: validatedParams,
             config: validatedConfig,
             secrets: validatedSecrets,
+            isEphemeral,
           });
         } catch (err) {
           rawResult = {

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

@@ -125,6 +125,7 @@ test('executes the task by calling the executor with proper parameters', async (
 
   expect(mockedActionExecutor.execute).toHaveBeenCalledWith({
     actionId: '2',
+    isEphemeral: false,
     params: { baz: true },
     relatedSavedObjects: [],
     request: expect.objectContaining({
@@ -250,6 +251,7 @@ test('uses API key when provided', async () => {
 
   expect(mockedActionExecutor.execute).toHaveBeenCalledWith({
     actionId: '2',
+    isEphemeral: false,
     params: { baz: true },
     relatedSavedObjects: [],
     request: expect.objectContaining({
@@ -293,6 +295,7 @@ test('uses relatedSavedObjects when provided', async () => {
 
   expect(mockedActionExecutor.execute).toHaveBeenCalledWith({
     actionId: '2',
+    isEphemeral: false,
     params: { baz: true },
     relatedSavedObjects: [
       {
@@ -334,14 +337,15 @@ test('sanitizes invalid relatedSavedObjects when provided', async () => {
   await taskRunner.run();
   expect(mockedActionExecutor.execute).toHaveBeenCalledWith({
     actionId: '2',
+    isEphemeral: false,
     params: { baz: true },
-    relatedSavedObjects: [],
     request: expect.objectContaining({
       headers: {
         // base64 encoded "123:abc"
         authorization: 'ApiKey MTIzOmFiYw==',
       },
     }),
+    relatedSavedObjects: [],
     taskInfo: {
       scheduled: new Date(),
     },
@@ -369,6 +373,7 @@ test(`doesn't use API key when not provided`, async () => {
 
   expect(mockedActionExecutor.execute).toHaveBeenCalledWith({
     actionId: '2',
+    isEphemeral: false,
     params: { baz: true },
     relatedSavedObjects: [],
     request: expect.objectContaining({