github-mirrors/kibana
1
0
Fork 0
mirror of https://github.com/elastic/kibana.git synced 2025-04-24 17:59:23 -04:00
Code Issues Projects Releases Packages Wiki Activity

Change Alerts > Actions execution order (#143577)

Browse source 
* Change Alerts > Actions execution order
This commit is contained in:
Ersin Erdal 2022-11-07 16:49:31 +01:00 • committed by GitHub
parent c78b7fad9b
commit ad8e48f87c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 1192 additions and 1296 deletions
Download patch file Download diff file Expand all files Collapse all files

595
x-pack/plugins/alerting/server/task_runner/create_execution_handler.test.ts
View file

@ -1,595 +0,0 @@
/*
* 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 { createExecutionHandler } from './create_execution_handler';
import { CreateExecutionHandlerOptions } from './types';
import { loggingSystemMock } from '@kbn/core/server/mocks';
import {
actionsClientMock,
actionsMock,
renderActionParameterTemplatesDefault,
} from '@kbn/actions-plugin/server/mocks';
import { KibanaRequest } from '@kbn/core/server';
import { asSavedObjectExecutionSource } from '@kbn/actions-plugin/server';
import { InjectActionParamsOpts } from './inject_action_params';
import { NormalizedRuleType } from '../rule_type_registry';
import {
ActionsCompletion,
AlertInstanceContext,
AlertInstanceState,
RuleTypeParams,
RuleTypeState,
} from '../types';
import { RuleRunMetricsStore } from '../lib/rule_run_metrics_store';
import { alertingEventLoggerMock } from '../lib/alerting_event_logger/alerting_event_logger.mock';
jest.mock('./inject_action_params', () => ({
injectActionParams: jest.fn(),
}));
const alertingEventLogger = alertingEventLoggerMock.create();
const ruleType: NormalizedRuleType<
RuleTypeParams,
RuleTypeParams,
RuleTypeState,
AlertInstanceState,
AlertInstanceContext,
'default' | 'other-group',
'recovered'
> = {
id: 'test',
name: 'Test',
actionGroups: [
{ id: 'default', name: 'Default' },
{ id: 'other-group', name: 'Other Group' },
],
defaultActionGroupId: 'default',
minimumLicenseRequired: 'basic',
isExportable: true,
recoveryActionGroup: {
id: 'recovered',
name: 'Recovered',
},
executor: jest.fn(),
producer: 'alerts',
};
const actionsClient = actionsClientMock.create();
const mockActionsPlugin = actionsMock.createStart();
const createExecutionHandlerParams: jest.Mocked<
CreateExecutionHandlerOptions<
RuleTypeParams,
RuleTypeParams,
RuleTypeState,
AlertInstanceState,
AlertInstanceContext,
'default' | 'other-group',
'recovered'
>
> = {
actionsPlugin: mockActionsPlugin,
spaceId: 'test1',
ruleId: '1',
ruleName: 'name-of-alert',
ruleConsumer: 'rule-consumer',
executionId: '5f6aa57d-3e22-484e-bae8-cbed868f4d28',
tags: ['tag-A', 'tag-B'],
apiKey: 'MTIzOmFiYw==',
kibanaBaseUrl: 'http://localhost:5601',
ruleType,
logger: loggingSystemMock.create().get(),
alertingEventLogger,
actions: [
{
id: '1',
group: 'default',
actionTypeId: 'test',
params: {
foo: true,
contextVal: 'My {{context.value}} goes here',
stateVal: 'My {{state.value}} goes here',
alertVal: 'My {{alertId}} {{alertName}} {{spaceId}} {{tags}} {{alertInstanceId}} goes here',
},
},
],
request: {} as KibanaRequest,
ruleParams: {
foo: true,
contextVal: 'My other {{context.value}} goes here',
stateVal: 'My other {{state.value}} goes here',
},
supportsEphemeralTasks: false,
maxEphemeralActionsPerRule: 10,
actionsConfigMap: {
default: {
max: 1000,
},
},
};
let ruleRunMetricsStore: RuleRunMetricsStore;
describe('Create Execution Handler', () => {
beforeEach(() => {
jest.resetAllMocks();
jest
.requireMock('./inject_action_params')
.injectActionParams.mockImplementation(
({ actionParams }: InjectActionParamsOpts) => actionParams
);
mockActionsPlugin.isActionTypeEnabled.mockReturnValue(true);
mockActionsPlugin.isActionExecutable.mockReturnValue(true);
mockActionsPlugin.getActionsClientWithRequest.mockResolvedValue(actionsClient);
mockActionsPlugin.renderActionParameterTemplates.mockImplementation(
renderActionParameterTemplatesDefault
);
ruleRunMetricsStore = new RuleRunMetricsStore();
});
test('enqueues execution per selected action', async () => {
const executionHandler = createExecutionHandler(createExecutionHandlerParams);
await executionHandler({
actionGroup: 'default',
state: {},
context: {},
alertId: '2',
ruleRunMetricsStore,
});
expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(1);
expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(1);
expect(mockActionsPlugin.getActionsClientWithRequest).toHaveBeenCalledWith(
createExecutionHandlerParams.request
);
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
expect(actionsClient.bulkEnqueueExecution.mock.calls[0]).toMatchInlineSnapshot(`
Array [
Array [
Object {
"apiKey": "MTIzOmFiYw==",
"consumer": "rule-consumer",
"executionId": "5f6aa57d-3e22-484e-bae8-cbed868f4d28",
"id": "1",
"params": Object {
"alertVal": "My 1 name-of-alert test1 tag-A,tag-B 2 goes here",
"contextVal": "My goes here",
"foo": true,
"stateVal": "My goes here",
},
"relatedSavedObjects": Array [
Object {
"id": "1",
"namespace": "test1",
"type": "alert",
"typeId": "test",
},
],
"source": Object {
"source": Object {
"id": "1",
"type": "alert",
},
"type": "SAVED_OBJECT",
},
"spaceId": "test1",
},
],
]
`);
expect(alertingEventLogger.logAction).toHaveBeenCalledTimes(1);
expect(alertingEventLogger.logAction).toHaveBeenNthCalledWith(1, {
id: '1',
typeId: 'test',
alertId: '2',
alertGroup: 'default',
});
expect(jest.requireMock('./inject_action_params').injectActionParams).toHaveBeenCalledWith({
ruleId: '1',
spaceId: 'test1',
actionTypeId: 'test',
actionParams: {
alertVal: 'My 1 name-of-alert test1 tag-A,tag-B 2 goes here',
contextVal: 'My goes here',
foo: true,
stateVal: 'My goes here',
},
});
expect(ruleRunMetricsStore.getTriggeredActionsStatus()).toBe(ActionsCompletion.COMPLETE);
});
test(`doesn't call actionsPlugin.execute for disabled actionTypes`, async () => {
// Mock two calls, one for check against actions[0] and the second for actions[1]
mockActionsPlugin.isActionExecutable.mockReturnValueOnce(false);
mockActionsPlugin.isActionTypeEnabled.mockReturnValueOnce(false);
mockActionsPlugin.isActionTypeEnabled.mockReturnValueOnce(true);
const executionHandler = createExecutionHandler({
...createExecutionHandlerParams,
actions: [
{
id: '2',
group: 'default',
actionTypeId: 'test2',
params: {
foo: true,
contextVal: 'My other {{context.value}} goes here',
stateVal: 'My other {{state.value}} goes here',
},
},
{
id: '2',
group: 'default',
actionTypeId: 'test2',
params: {
foo: true,
contextVal: 'My other {{context.value}} goes here',
stateVal: 'My other {{state.value}} goes here',
},
},
],
});
await executionHandler({
actionGroup: 'default',
state: {},
context: {},
alertId: '2',
ruleRunMetricsStore,
});
expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(1);
expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(2);
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledWith([
{
consumer: 'rule-consumer',
id: '2',
params: {
foo: true,
contextVal: 'My other goes here',
stateVal: 'My other goes here',
},
source: asSavedObjectExecutionSource({
id: '1',
type: 'alert',
}),
relatedSavedObjects: [
{
id: '1',
namespace: 'test1',
type: 'alert',
typeId: 'test',
},
],
spaceId: 'test1',
apiKey: createExecutionHandlerParams.apiKey,
executionId: '5f6aa57d-3e22-484e-bae8-cbed868f4d28',
},
]);
});
test('trow error error message when action type is disabled', async () => {
mockActionsPlugin.preconfiguredActions = [];
mockActionsPlugin.isActionExecutable.mockReturnValue(false);
mockActionsPlugin.isActionTypeEnabled.mockReturnValue(false);
const executionHandler = createExecutionHandler({
...createExecutionHandlerParams,
actions: [
{
id: '1',
group: 'default',
actionTypeId: '.slack',
params: {
foo: true,
},
},
{
id: '2',
group: 'default',
actionTypeId: '.slack',
params: {
foo: true,
contextVal: 'My other {{context.value}} goes here',
stateVal: 'My other {{state.value}} goes here',
},
},
],
});
await executionHandler({
actionGroup: 'default',
state: {},
context: {},
alertId: '2',
ruleRunMetricsStore,
});
expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(0);
expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(2);
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(0);
mockActionsPlugin.isActionExecutable.mockImplementation(() => true);
const executionHandlerForPreconfiguredAction = createExecutionHandler({
...createExecutionHandlerParams,
actions: [...createExecutionHandlerParams.actions],
});
await executionHandlerForPreconfiguredAction({
actionGroup: 'default',
state: {},
context: {},
alertId: '2',
ruleRunMetricsStore,
});
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
});
test('limits actionsPlugin.execute per action group', async () => {
const executionHandler = createExecutionHandler(createExecutionHandlerParams);
await executionHandler({
actionGroup: 'other-group',
state: {},
context: {},
alertId: '2',
ruleRunMetricsStore,
});
expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(0);
expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(0);
expect(actionsClient.bulkEnqueueExecution).not.toHaveBeenCalled();
});
test('context attribute gets parameterized', async () => {
const executionHandler = createExecutionHandler(createExecutionHandlerParams);
await executionHandler({
actionGroup: 'default',
context: { value: 'context-val' },
state: {},
alertId: '2',
ruleRunMetricsStore,
});
expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(1);
expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(1);
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
expect(actionsClient.bulkEnqueueExecution.mock.calls[0]).toMatchInlineSnapshot(`
Array [
Array [
Object {
"apiKey": "MTIzOmFiYw==",
"consumer": "rule-consumer",
"executionId": "5f6aa57d-3e22-484e-bae8-cbed868f4d28",
"id": "1",
"params": Object {
"alertVal": "My 1 name-of-alert test1 tag-A,tag-B 2 goes here",
"contextVal": "My context-val goes here",
"foo": true,
"stateVal": "My goes here",
},
"relatedSavedObjects": Array [
Object {
"id": "1",
"namespace": "test1",
"type": "alert",
"typeId": "test",
},
],
"source": Object {
"source": Object {
"id": "1",
"type": "alert",
},
"type": "SAVED_OBJECT",
},
"spaceId": "test1",
},
],
]
`);
});
test('state attribute gets parameterized', async () => {
const executionHandler = createExecutionHandler(createExecutionHandlerParams);
await executionHandler({
actionGroup: 'default',
context: {},
state: { value: 'state-val' },
alertId: '2',
ruleRunMetricsStore,
});
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
expect(actionsClient.bulkEnqueueExecution.mock.calls[0]).toMatchInlineSnapshot(`
Array [
Array [
Object {
"apiKey": "MTIzOmFiYw==",
"consumer": "rule-consumer",
"executionId": "5f6aa57d-3e22-484e-bae8-cbed868f4d28",
"id": "1",
"params": Object {
"alertVal": "My 1 name-of-alert test1 tag-A,tag-B 2 goes here",
"contextVal": "My goes here",
"foo": true,
"stateVal": "My state-val goes here",
},
"relatedSavedObjects": Array [
Object {
"id": "1",
"namespace": "test1",
"type": "alert",
"typeId": "test",
},
],
"source": Object {
"source": Object {
"id": "1",
"type": "alert",
},
"type": "SAVED_OBJECT",
},
"spaceId": "test1",
},
],
]
`);
});
test(`logs an error when action group isn't part of actionGroups available for the ruleType`, async () => {
const executionHandler = createExecutionHandler(createExecutionHandlerParams);
await executionHandler({
// we have to trick the compiler as this is an invalid type and this test checks whether we
// enforce this at runtime as well as compile time
actionGroup: 'invalid-group' as 'default' | 'other-group',
context: {},
state: {},
alertId: '2',
ruleRunMetricsStore,
});
expect(createExecutionHandlerParams.logger.error).toHaveBeenCalledWith(
'Invalid action group "invalid-group" for rule "test".'
);
expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(0);
expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(0);
expect(ruleRunMetricsStore.getTriggeredActionsStatus()).toBe(ActionsCompletion.COMPLETE);
});
test('Stops triggering actions when the number of total triggered actions is reached the number of max executable actions', async () => {
const executionHandler = createExecutionHandler({
...createExecutionHandlerParams,
actionsConfigMap: {
default: {
max: 2,
},
},
actions: [
{
id: '1',
group: 'default',
actionTypeId: 'test2',
params: {
foo: true,
contextVal: 'My other {{context.value}} goes here',
stateVal: 'My other {{state.value}} goes here',
},
},
{
id: '2',
group: 'default',
actionTypeId: 'test2',
params: {
foo: true,
contextVal: 'My other {{context.value}} goes here',
stateVal: 'My other {{state.value}} goes here',
},
},
{
id: '3',
group: 'default',
actionTypeId: 'test3',
params: {
foo: true,
contextVal: '{{context.value}} goes here',
stateVal: '{{state.value}} goes here',
},
},
],
});
ruleRunMetricsStore = new RuleRunMetricsStore();
await executionHandler({
actionGroup: 'default',
context: {},
state: { value: 'state-val' },
alertId: '2',
ruleRunMetricsStore,
});
expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(2);
expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(3);
expect(ruleRunMetricsStore.getTriggeredActionsStatus()).toBe(ActionsCompletion.PARTIAL);
expect(createExecutionHandlerParams.logger.debug).toHaveBeenCalledTimes(1);
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
});
test('Skips triggering actions for a specific action type when it reaches the limit for that specific action type', async () => {
const executionHandler = createExecutionHandler({
...createExecutionHandlerParams,
actionsConfigMap: {
default: {
max: 4,
},
'test-action-type-id': {
max: 1,
},
},
actions: [
...createExecutionHandlerParams.actions,
{
id: '2',
group: 'default',
actionTypeId: 'test-action-type-id',
params: {
foo: true,
contextVal: 'My other {{context.value}} goes here',
stateVal: 'My other {{state.value}} goes here',
},
},
{
id: '3',
group: 'default',
actionTypeId: 'test-action-type-id',
params: {
foo: true,
contextVal: '{{context.value}} goes here',
stateVal: '{{state.value}} goes here',
},
},
{
id: '4',
group: 'default',
actionTypeId: 'another-action-type-id',
params: {
foo: true,
contextVal: '{{context.value}} goes here',
stateVal: '{{state.value}} goes here',
},
},
{
id: '5',
group: 'default',
actionTypeId: 'another-action-type-id',
params: {
foo: true,
contextVal: '{{context.value}} goes here',
stateVal: '{{state.value}} goes here',
},
},
],
});
ruleRunMetricsStore = new RuleRunMetricsStore();
await executionHandler({
actionGroup: 'default',
context: {},
state: { value: 'state-val' },
alertId: '2',
ruleRunMetricsStore,
});
expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(4);
expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(5);
expect(ruleRunMetricsStore.getStatusByConnectorType('test').numberOfTriggeredActions).toBe(1);
expect(
ruleRunMetricsStore.getStatusByConnectorType('test-action-type-id').numberOfTriggeredActions
).toBe(1);
expect(
ruleRunMetricsStore.getStatusByConnectorType('another-action-type-id')
.numberOfTriggeredActions
).toBe(2);
expect(ruleRunMetricsStore.getTriggeredActionsStatus()).toBe(ActionsCompletion.PARTIAL);
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
});
});

215
x-pack/plugins/alerting/server/task_runner/create_execution_handler.ts
View file

@ -1,215 +0,0 @@
/*
* 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 { asSavedObjectExecutionSource } from '@kbn/actions-plugin/server';
import { isEphemeralTaskRejectedDueToCapacityError } from '@kbn/task-manager-plugin/server';
import { chunk } from 'lodash';
import { transformActionParams } from './transform_action_params';
import { injectActionParams } from './inject_action_params';
import {
ActionsCompletion,
AlertInstanceContext,
AlertInstanceState,
RuleTypeParams,
RuleTypeState,
} from '../types';
import { CreateExecutionHandlerOptions, ExecutionHandlerOptions } from './types';
export type ExecutionHandler<ActionGroupIds extends string> = (
options: ExecutionHandlerOptions<ActionGroupIds>
) => Promise<void>;
export function createExecutionHandler<
Params extends RuleTypeParams,
ExtractedParams extends RuleTypeParams,
State extends RuleTypeState,
InstanceState extends AlertInstanceState,
InstanceContext extends AlertInstanceContext,
ActionGroupIds extends string,
RecoveryActionGroupId extends string
>({
logger,
ruleId,
ruleName,
ruleConsumer,
executionId,
tags,
actionsPlugin,
actions: ruleActions,
spaceId,
apiKey,
ruleType,
kibanaBaseUrl,
alertingEventLogger,
request,
ruleParams,
supportsEphemeralTasks,
maxEphemeralActionsPerRule,
actionsConfigMap,
}: CreateExecutionHandlerOptions<
Params,
ExtractedParams,
State,
InstanceState,
InstanceContext,
ActionGroupIds,
RecoveryActionGroupId
>): ExecutionHandler<ActionGroupIds | RecoveryActionGroupId> {
const ruleTypeActionGroups = new Map(
ruleType.actionGroups.map((actionGroup) => [actionGroup.id, actionGroup.name])
);
const CHUNK_SIZE = 1000;
return async ({
actionGroup,
context,
state,
ruleRunMetricsStore,
alertId,
}: ExecutionHandlerOptions<ActionGroupIds | RecoveryActionGroupId>) => {
if (!ruleTypeActionGroups.has(actionGroup)) {
logger.error(`Invalid action group "${actionGroup}" for rule "${ruleType.id}".`);
return;
}
const actions = ruleActions
.filter(({ group }) => group === actionGroup)
.map((action) => {
return {
...action,
params: transformActionParams({
actionsPlugin,
alertId: ruleId,
alertType: ruleType.id,
actionTypeId: action.actionTypeId,
alertName: ruleName,
spaceId,
tags,
alertInstanceId: alertId,
alertActionGroup: actionGroup,
alertActionGroupName: ruleTypeActionGroups.get(actionGroup)!,
context,
actionParams: action.params,
actionId: action.id,
state,
kibanaBaseUrl,
alertParams: ruleParams,
}),
};
})
.map((action) => ({
...action,
params: injectActionParams({
ruleId,
spaceId,
actionParams: action.params,
actionTypeId: action.actionTypeId,
}),
}));
ruleRunMetricsStore.incrementNumberOfGeneratedActions(actions.length);
const actionsClient = await actionsPlugin.getActionsClientWithRequest(request);
let ephemeralActionsToSchedule = maxEphemeralActionsPerRule;
const bulkActions = [];
const logActions = [];
for (const action of actions) {
const { actionTypeId } = action;
ruleRunMetricsStore.incrementNumberOfGeneratedActionsByConnectorType(actionTypeId);
if (ruleRunMetricsStore.hasReachedTheExecutableActionsLimit(actionsConfigMap)) {
ruleRunMetricsStore.setTriggeredActionsStatusByConnectorType({
actionTypeId,
status: ActionsCompletion.PARTIAL,
});
logger.debug(
`Rule "${ruleId}" skipped scheduling action "${action.id}" because the maximum number of allowed actions has been reached.`
);
break;
}
if (
ruleRunMetricsStore.hasReachedTheExecutableActionsLimitByConnectorType({
actionTypeId,
actionsConfigMap,
})
) {
if (!ruleRunMetricsStore.hasConnectorTypeReachedTheLimit(actionTypeId)) {
logger.debug(
`Rule "${ruleId}" skipped scheduling action "${action.id}" because the maximum number of allowed actions for connector type ${actionTypeId} has been reached.`
);
}
ruleRunMetricsStore.setTriggeredActionsStatusByConnectorType({
actionTypeId,
status: ActionsCompletion.PARTIAL,
});
continue;
}
if (!actionsPlugin.isActionExecutable(action.id, actionTypeId, { notifyUsage: true })) {
logger.warn(
`Rule "${ruleId}" skipped scheduling action "${action.id}" because it is disabled`
);
continue;
}
ruleRunMetricsStore.incrementNumberOfTriggeredActions();
ruleRunMetricsStore.incrementNumberOfTriggeredActionsByConnectorType(actionTypeId);
const namespace = spaceId === 'default' ? {} : { namespace: spaceId };
const enqueueOptions = {
id: action.id,
params: action.params,
spaceId,
apiKey: apiKey ?? null,
consumer: ruleConsumer,
source: asSavedObjectExecutionSource({
id: ruleId,
type: 'alert',
}),
executionId,
relatedSavedObjects: [
{
id: ruleId,
type: 'alert',
namespace: namespace.namespace,
typeId: ruleType.id,
},
],
};
if (supportsEphemeralTasks && ephemeralActionsToSchedule > 0) {
ephemeralActionsToSchedule--;
try {
await actionsClient.ephemeralEnqueuedExecution(enqueueOptions);
} catch (err) {
if (isEphemeralTaskRejectedDueToCapacityError(err)) {
bulkActions.push(enqueueOptions);
}
}
} else {
bulkActions.push(enqueueOptions);
}
logActions.push({
id: action.id,
typeId: actionTypeId,
alertId,
alertGroup: actionGroup,
});
}
for (const c of chunk(bulkActions, CHUNK_SIZE)) {
await actionsClient.bulkEnqueueExecution(c);
}
for (const action of logActions) {
alertingEventLogger.logAction(action);
}
};
}

732
x-pack/plugins/alerting/server/task_runner/execution_handler.test.ts Normal file
View file

@ -0,0 +1,732 @@
/*
* 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 { ExecutionHandler } from './execution_handler';
import { loggingSystemMock } from '@kbn/core/server/mocks';
import {
actionsClientMock,
actionsMock,
renderActionParameterTemplatesDefault,
} from '@kbn/actions-plugin/server/mocks';
import { KibanaRequest } from '@kbn/core/server';
import { InjectActionParamsOpts } from './inject_action_params';
import { NormalizedRuleType } from '../rule_type_registry';
import { ActionsCompletion, RuleTypeParams, RuleTypeState, SanitizedRule } from '../types';
import { RuleRunMetricsStore } from '../lib/rule_run_metrics_store';
import { alertingEventLoggerMock } from '../lib/alerting_event_logger/alerting_event_logger.mock';
import { TaskRunnerContext } from './task_runner_factory';
import { ConcreteTaskInstance } from '@kbn/task-manager-plugin/server';
import { Alert } from '../alert';
import { AlertInstanceState, AlertInstanceContext } from '../../common';
import { asSavedObjectExecutionSource } from '@kbn/actions-plugin/server';
import sinon from 'sinon';
jest.mock('./inject_action_params', () => ({
injectActionParams: jest.fn(),
}));
const alertingEventLogger = alertingEventLoggerMock.create();
const actionsClient = actionsClientMock.create();
const mockActionsPlugin = actionsMock.createStart();
const apiKey = Buffer.from('123:abc').toString('base64');
const ruleType: NormalizedRuleType<
RuleTypeParams,
RuleTypeParams,
RuleTypeState,
AlertInstanceState,
AlertInstanceContext,
'default' | 'other-group',
'recovered'
> = {
id: 'test',
name: 'Test',
actionGroups: [
{ id: 'default', name: 'Default' },
{ id: 'recovered', name: 'Recovered' },
{ id: 'other-group', name: 'Other Group' },
],
defaultActionGroupId: 'default',
minimumLicenseRequired: 'basic',
isExportable: true,
recoveryActionGroup: {
id: 'recovered',
name: 'Recovered',
},
executor: jest.fn(),
producer: 'alerts',
};
const rule = {
id: '1',
name: 'name-of-alert',
tags: ['tag-A', 'tag-B'],
mutedInstanceIds: [],
params: {
foo: true,
contextVal: 'My other {{context.value}} goes here',
stateVal: 'My other {{state.value}} goes here',
},
actions: [
{
id: '1',
group: 'default',
actionTypeId: 'test',
params: {
foo: true,
contextVal: 'My {{context.value}} goes here',
stateVal: 'My {{state.value}} goes here',
alertVal: 'My {{alertId}} {{alertName}} {{spaceId}} {{tags}} {{alertInstanceId}} goes here',
},
},
],
} as unknown as SanitizedRule<RuleTypeParams>;
const defaultExecutionParams = {
rule,
ruleType,
logger: loggingSystemMock.create().get(),
taskRunnerContext: {
actionsConfigMap: {
default: {
max: 1000,
},
},
actionsPlugin: mockActionsPlugin,
} as unknown as TaskRunnerContext,
apiKey,
ruleConsumer: 'rule-consumer',
executionId: '5f6aa57d-3e22-484e-bae8-cbed868f4d28',
ruleLabel: 'rule-label',
request: {} as KibanaRequest,
alertingEventLogger,
taskInstance: {
params: { spaceId: 'test1', alertId: '1' },
} as unknown as ConcreteTaskInstance,
actionsClient,
};
let ruleRunMetricsStore: RuleRunMetricsStore;
let clock: sinon.SinonFakeTimers;
type ActionGroup = 'default' | 'other-group' | 'recovered';
const generateAlert = ({
id,
group = 'default',
context,
state,
scheduleActions = true,
}: {
id: number;
group?: ActionGroup;
context?: AlertInstanceContext;
state?: AlertInstanceState;
scheduleActions?: boolean;
}) => {
const alert = new Alert<
AlertInstanceState,
AlertInstanceContext,
'default' | 'other-group' | 'recovered'
>(String(id), {
state: state || { test: true },
meta: {
lastScheduledActions: {
date: new Date(),
group,
},
},
});
if (scheduleActions) {
alert.scheduleActions(group);
}
if (context) {
alert.setContext(context);
}
return { [id]: alert };
};
// @ts-ignore
const generateExecutionParams = (params = {}) => {
return {
...defaultExecutionParams,
...params,
ruleRunMetricsStore,
};
};
describe('Execution Handler', () => {
beforeEach(() => {
jest.resetAllMocks();
jest
.requireMock('./inject_action_params')
.injectActionParams.mockImplementation(
({ actionParams }: InjectActionParamsOpts) => actionParams
);
mockActionsPlugin.isActionTypeEnabled.mockReturnValue(true);
mockActionsPlugin.isActionExecutable.mockReturnValue(true);
mockActionsPlugin.getActionsClientWithRequest.mockResolvedValue(actionsClient);
mockActionsPlugin.renderActionParameterTemplates.mockImplementation(
renderActionParameterTemplatesDefault
);
ruleRunMetricsStore = new RuleRunMetricsStore();
});
beforeAll(() => {
clock = sinon.useFakeTimers();
});
afterAll(() => clock.restore());
test('enqueues execution per selected action', async () => {
const executionHandler = new ExecutionHandler(generateExecutionParams());
await executionHandler.run(generateAlert({ id: 1 }));
expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(1);
expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(1);
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
expect(actionsClient.bulkEnqueueExecution.mock.calls[0]).toMatchInlineSnapshot(`
Array [
Array [
Object {
"apiKey": "MTIzOmFiYw==",
"consumer": "rule-consumer",
"executionId": "5f6aa57d-3e22-484e-bae8-cbed868f4d28",
"id": "1",
"params": Object {
"alertVal": "My 1 name-of-alert test1 tag-A,tag-B 1 goes here",
"contextVal": "My goes here",
"foo": true,
"stateVal": "My goes here",
},
"relatedSavedObjects": Array [
Object {
"id": "1",
"namespace": "test1",
"type": "alert",
"typeId": "test",
},
],
"source": Object {
"source": Object {
"id": "1",
"type": "alert",
},
"type": "SAVED_OBJECT",
},
"spaceId": "test1",
},
],
]
`);
expect(alertingEventLogger.logAction).toHaveBeenCalledTimes(1);
expect(alertingEventLogger.logAction).toHaveBeenNthCalledWith(1, {
id: '1',
typeId: 'test',
alertId: '1',
alertGroup: 'default',
});
expect(jest.requireMock('./inject_action_params').injectActionParams).toHaveBeenCalledWith({
ruleId: '1',
spaceId: 'test1',
actionTypeId: 'test',
actionParams: {
alertVal: 'My 1 name-of-alert test1 tag-A,tag-B 1 goes here',
contextVal: 'My goes here',
foo: true,
stateVal: 'My goes here',
},
});
expect(ruleRunMetricsStore.getTriggeredActionsStatus()).toBe(ActionsCompletion.COMPLETE);
});
test(`doesn't call actionsPlugin.execute for disabled actionTypes`, async () => {
// Mock two calls, one for check against actions[0] and the second for actions[1]
mockActionsPlugin.isActionExecutable.mockReturnValueOnce(false);
mockActionsPlugin.isActionTypeEnabled.mockReturnValueOnce(false);
mockActionsPlugin.isActionTypeEnabled.mockReturnValueOnce(true);
const executionHandler = new ExecutionHandler(
generateExecutionParams({
rule: {
...defaultExecutionParams.rule,
actions: [
{
id: '2',
group: 'default',
actionTypeId: 'test2',
params: {
foo: true,
contextVal: 'My other {{context.value}} goes here',
stateVal: 'My other {{state.value}} goes here',
},
},
{
id: '2',
group: 'default',
actionTypeId: 'test2',
params: {
foo: true,
contextVal: 'My other {{context.value}} goes here',
stateVal: 'My other {{state.value}} goes here',
},
},
],
},
})
);
await executionHandler.run(generateAlert({ id: 1 }));
expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(1);
expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(2);
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledWith([
{
consumer: 'rule-consumer',
id: '2',
params: {
foo: true,
contextVal: 'My other goes here',
stateVal: 'My other goes here',
},
source: asSavedObjectExecutionSource({
id: '1',
type: 'alert',
}),
relatedSavedObjects: [
{
id: '1',
namespace: 'test1',
type: 'alert',
typeId: 'test',
},
],
spaceId: 'test1',
apiKey,
executionId: '5f6aa57d-3e22-484e-bae8-cbed868f4d28',
},
]);
});
test('trow error error message when action type is disabled', async () => {
mockActionsPlugin.preconfiguredActions = [];
mockActionsPlugin.isActionExecutable.mockReturnValue(false);
mockActionsPlugin.isActionTypeEnabled.mockReturnValue(false);
const executionHandler = new ExecutionHandler(
generateExecutionParams({
rule: {
...defaultExecutionParams.rule,
actions: [
{
id: '1',
group: 'default',
actionTypeId: '.slack',
params: {
foo: true,
},
},
{
id: '2',
group: 'default',
actionTypeId: '.slack',
params: {
foo: true,
contextVal: 'My other {{context.value}} goes here',
stateVal: 'My other {{state.value}} goes here',
},
},
],
},
})
);
await executionHandler.run(generateAlert({ id: 2 }));
expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(0);
expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(2);
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(0);
mockActionsPlugin.isActionExecutable.mockImplementation(() => true);
const executionHandlerForPreconfiguredAction = new ExecutionHandler({
...defaultExecutionParams,
ruleRunMetricsStore,
});
await executionHandlerForPreconfiguredAction.run(generateAlert({ id: 2 }));
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
});
test('limits actionsPlugin.execute per action group', async () => {
const executionHandler = new ExecutionHandler(generateExecutionParams());
await executionHandler.run(generateAlert({ id: 2, group: 'other-group' }));
expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(0);
expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(0);
expect(actionsClient.bulkEnqueueExecution).not.toHaveBeenCalled();
});
test('context attribute gets parameterized', async () => {
const executionHandler = new ExecutionHandler(generateExecutionParams());
await executionHandler.run(generateAlert({ id: 2, context: { value: 'context-val' } }));
expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(1);
expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(1);
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
expect(actionsClient.bulkEnqueueExecution.mock.calls[0]).toMatchInlineSnapshot(`
Array [
Array [
Object {
"apiKey": "MTIzOmFiYw==",
"consumer": "rule-consumer",
"executionId": "5f6aa57d-3e22-484e-bae8-cbed868f4d28",
"id": "1",
"params": Object {
"alertVal": "My 1 name-of-alert test1 tag-A,tag-B 2 goes here",
"contextVal": "My context-val goes here",
"foo": true,
"stateVal": "My goes here",
},
"relatedSavedObjects": Array [
Object {
"id": "1",
"namespace": "test1",
"type": "alert",
"typeId": "test",
},
],
"source": Object {
"source": Object {
"id": "1",
"type": "alert",
},
"type": "SAVED_OBJECT",
},
"spaceId": "test1",
},
],
]
`);
});
test('state attribute gets parameterized', async () => {
const executionHandler = new ExecutionHandler(generateExecutionParams());
await executionHandler.run(generateAlert({ id: 2, state: { value: 'state-val' } }));
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
expect(actionsClient.bulkEnqueueExecution.mock.calls[0]).toMatchInlineSnapshot(`
Array [
Array [
Object {
"apiKey": "MTIzOmFiYw==",
"consumer": "rule-consumer",
"executionId": "5f6aa57d-3e22-484e-bae8-cbed868f4d28",
"id": "1",
"params": Object {
"alertVal": "My 1 name-of-alert test1 tag-A,tag-B 2 goes here",
"contextVal": "My goes here",
"foo": true,
"stateVal": "My state-val goes here",
},
"relatedSavedObjects": Array [
Object {
"id": "1",
"namespace": "test1",
"type": "alert",
"typeId": "test",
},
],
"source": Object {
"source": Object {
"id": "1",
"type": "alert",
},
"type": "SAVED_OBJECT",
},
"spaceId": "test1",
},
],
]
`);
});
test(`logs an error when action group isn't part of actionGroups available for the ruleType`, async () => {
const executionHandler = new ExecutionHandler(generateExecutionParams());
await executionHandler.run(
generateAlert({ id: 2, group: 'invalid-group' as 'default' | 'other-group' })
);
expect(defaultExecutionParams.logger.error).toHaveBeenCalledWith(
'Invalid action group "invalid-group" for rule "test".'
);
expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(0);
expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(0);
expect(ruleRunMetricsStore.getTriggeredActionsStatus()).toBe(ActionsCompletion.COMPLETE);
});
test('Stops triggering actions when the number of total triggered actions is reached the number of max executable actions', async () => {
const executionHandler = new ExecutionHandler(
generateExecutionParams({
...defaultExecutionParams,
taskRunnerContext: {
...defaultExecutionParams.taskRunnerContext,
actionsConfigMap: {
default: {
max: 2,
},
},
},
rule: {
...defaultExecutionParams.rule,
actions: [
{
id: '1',
group: 'default',
actionTypeId: 'test2',
params: {
foo: true,
contextVal: 'My other {{context.value}} goes here',
stateVal: 'My other {{state.value}} goes here',
},
},
{
id: '2',
group: 'default',
actionTypeId: 'test2',
params: {
foo: true,
contextVal: 'My other {{context.value}} goes here',
stateVal: 'My other {{state.value}} goes here',
},
},
{
id: '3',
group: 'default',
actionTypeId: 'test3',
params: {
foo: true,
contextVal: '{{context.value}} goes here',
stateVal: '{{state.value}} goes here',
},
},
],
},
})
);
await executionHandler.run(generateAlert({ id: 2, state: { value: 'state-val' } }));
expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(2);
expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(3);
expect(ruleRunMetricsStore.getTriggeredActionsStatus()).toBe(ActionsCompletion.PARTIAL);
expect(defaultExecutionParams.logger.debug).toHaveBeenCalledTimes(1);
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
});
test('Skips triggering actions for a specific action type when it reaches the limit for that specific action type', async () => {
const executionHandler = new ExecutionHandler(
generateExecutionParams({
...defaultExecutionParams,
taskRunnerContext: {
...defaultExecutionParams.taskRunnerContext,
actionsConfigMap: {
default: {
max: 4,
},
'test-action-type-id': {
max: 1,
},
},
},
rule: {
...defaultExecutionParams.rule,
actions: [
...defaultExecutionParams.rule.actions,
{
id: '2',
group: 'default',
actionTypeId: 'test-action-type-id',
params: {
foo: true,
contextVal: 'My other {{context.value}} goes here',
stateVal: 'My other {{state.value}} goes here',
},
},
{
id: '3',
group: 'default',
actionTypeId: 'test-action-type-id',
params: {
foo: true,
contextVal: '{{context.value}} goes here',
stateVal: '{{state.value}} goes here',
},
},
{
id: '4',
group: 'default',
actionTypeId: 'another-action-type-id',
params: {
foo: true,
contextVal: '{{context.value}} goes here',
stateVal: '{{state.value}} goes here',
},
},
{
id: '5',
group: 'default',
actionTypeId: 'another-action-type-id',
params: {
foo: true,
contextVal: '{{context.value}} goes here',
stateVal: '{{state.value}} goes here',
},
},
],
},
})
);
await executionHandler.run(generateAlert({ id: 2, state: { value: 'state-val' } }));
expect(ruleRunMetricsStore.getNumberOfTriggeredActions()).toBe(4);
expect(ruleRunMetricsStore.getNumberOfGeneratedActions()).toBe(5);
expect(ruleRunMetricsStore.getStatusByConnectorType('test').numberOfTriggeredActions).toBe(1);
expect(
ruleRunMetricsStore.getStatusByConnectorType('test-action-type-id').numberOfTriggeredActions
).toBe(1);
expect(
ruleRunMetricsStore.getStatusByConnectorType('another-action-type-id')
.numberOfTriggeredActions
).toBe(2);
expect(ruleRunMetricsStore.getTriggeredActionsStatus()).toBe(ActionsCompletion.PARTIAL);
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
});
test('schedules alerts with recovered actions', async () => {
const executionHandler = new ExecutionHandler(
generateExecutionParams({
...defaultExecutionParams,
rule: {
...defaultExecutionParams.rule,
actions: [
{
id: '1',
group: 'recovered',
actionTypeId: 'test',
params: {
foo: true,
contextVal: 'My {{context.value}} goes here',
stateVal: 'My {{state.value}} goes here',
alertVal:
'My {{alertId}} {{alertName}} {{spaceId}} {{tags}} {{alertInstanceId}} goes here',
},
},
],
},
})
);
await executionHandler.run(generateAlert({ id: 1, scheduleActions: false }), true);
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
expect(actionsClient.bulkEnqueueExecution.mock.calls[0]).toMatchInlineSnapshot(`
Array [
Array [
Object {
"apiKey": "MTIzOmFiYw==",
"consumer": "rule-consumer",
"executionId": "5f6aa57d-3e22-484e-bae8-cbed868f4d28",
"id": "1",
"params": Object {
"alertVal": "My 1 name-of-alert test1 tag-A,tag-B 1 goes here",
"contextVal": "My goes here",
"foo": true,
"stateVal": "My goes here",
},
"relatedSavedObjects": Array [
Object {
"id": "1",
"namespace": "test1",
"type": "alert",
"typeId": "test",
},
],
"source": Object {
"source": Object {
"id": "1",
"type": "alert",
},
"type": "SAVED_OBJECT",
},
"spaceId": "test1",
},
],
]
`);
});
test('does not schedule alerts with recovered actions that are muted', async () => {
const executionHandler = new ExecutionHandler(
generateExecutionParams({
...defaultExecutionParams,
rule: {
...defaultExecutionParams.rule,
mutedInstanceIds: ['1'],
actions: [
{
id: '1',
group: 'recovered',
actionTypeId: 'test',
params: {
foo: true,
contextVal: 'My {{context.value}} goes here',
stateVal: 'My {{state.value}} goes here',
alertVal:
'My {{alertId}} {{alertName}} {{spaceId}} {{tags}} {{alertInstanceId}} goes here',
},
},
],
},
})
);
await executionHandler.run(generateAlert({ id: 1, scheduleActions: false }), true);
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(0);
expect(defaultExecutionParams.logger.debug).nthCalledWith(
1,
`skipping scheduling of actions for '1' in rule ${defaultExecutionParams.ruleLabel}: rule is muted`
);
});
test('does not schedule active alerts that are throttled', async () => {
const executionHandler = new ExecutionHandler(
generateExecutionParams({
...defaultExecutionParams,
rule: {
...defaultExecutionParams.rule,
throttle: '1m',
},
})
);
await executionHandler.run(generateAlert({ id: 1 }));
clock.tick(30000);
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(0);
expect(defaultExecutionParams.logger.debug).nthCalledWith(
1,
`skipping scheduling of actions for '1' in rule ${defaultExecutionParams.ruleLabel}: rule is throttled`
);
});
test('does not schedule active alerts that are muted', async () => {
const executionHandler = new ExecutionHandler(
generateExecutionParams({
...defaultExecutionParams,
rule: {
...defaultExecutionParams.rule,
mutedInstanceIds: ['1'],
},
})
);
await executionHandler.run(generateAlert({ id: 1 }));
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(0);
expect(defaultExecutionParams.logger.debug).nthCalledWith(
1,
`skipping scheduling of actions for '1' in rule ${defaultExecutionParams.ruleLabel}: rule is muted`
);
});
});

410
x-pack/plugins/alerting/server/task_runner/execution_handler.ts Normal file
View file

@ -0,0 +1,410 @@
/*
* 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 type { PublicMethodsOf } from '@kbn/utility-types';
import { Logger } from '@kbn/core/server';
import { asSavedObjectExecutionSource } from '@kbn/actions-plugin/server';
import { isEphemeralTaskRejectedDueToCapacityError } from '@kbn/task-manager-plugin/server';
import { ExecuteOptions as EnqueueExecutionOptions } from '@kbn/actions-plugin/server/create_execute_function';
import { ActionsClient } from '@kbn/actions-plugin/server/actions_client';
import { chunk } from 'lodash';
import { AlertingEventLogger } from '../lib/alerting_event_logger/alerting_event_logger';
import { RawRule } from '../types';
import { RuleRunMetricsStore } from '../lib/rule_run_metrics_store';
import { injectActionParams } from './inject_action_params';
import { ExecutionHandlerOptions, RuleTaskInstance } from './types';
import { TaskRunnerContext } from './task_runner_factory';
import { transformActionParams } from './transform_action_params';
import { Alert } from '../alert';
import { NormalizedRuleType } from '../rule_type_registry';
import {
ActionsCompletion,
AlertInstanceContext,
AlertInstanceState,
RuleAction,
RuleTypeParams,
RuleTypeState,
SanitizedRule,
} from '../../common';
enum Reasons {
MUTED = 'muted',
THROTTLED = 'throttled',
ACTION_GROUP_NOT_CHANGED = 'actionGroupHasNotChanged',
}
export class ExecutionHandler<
Params extends RuleTypeParams,
ExtractedParams extends RuleTypeParams,
RuleState extends RuleTypeState,
State extends AlertInstanceState,
Context extends AlertInstanceContext,
ActionGroupIds extends string,
RecoveryActionGroupId extends string
> {
private logger: Logger;
private alertingEventLogger: PublicMethodsOf<AlertingEventLogger>;
private rule: SanitizedRule<Params>;
private ruleType: NormalizedRuleType<
Params,
ExtractedParams,
RuleState,
State,
Context,
ActionGroupIds,
RecoveryActionGroupId
>;
private taskRunnerContext: TaskRunnerContext;
private taskInstance: RuleTaskInstance;
private ruleRunMetricsStore: RuleRunMetricsStore;
private apiKey: RawRule['apiKey'];
private ruleConsumer: string;
private executionId: string;
private ruleLabel: string;
private ephemeralActionsToSchedule: number;
private CHUNK_SIZE = 1000;
private skippedAlerts: { [key: string]: { reason: string } } = {};
private actionsClient: PublicMethodsOf<ActionsClient>;
private ruleTypeActionGroups?: Map<ActionGroupIds | RecoveryActionGroupId, string>;
private mutedAlertIdsSet?: Set<string>;
constructor({
rule,
ruleType,
logger,
alertingEventLogger,
taskRunnerContext,
taskInstance,
ruleRunMetricsStore,
apiKey,
ruleConsumer,
executionId,
ruleLabel,
actionsClient,
}: ExecutionHandlerOptions<
Params,
ExtractedParams,
RuleState,
State,
Context,
ActionGroupIds,
RecoveryActionGroupId
>) {
this.logger = logger;
this.alertingEventLogger = alertingEventLogger;
this.rule = rule;
this.ruleType = ruleType;
this.taskRunnerContext = taskRunnerContext;
this.taskInstance = taskInstance;
this.ruleRunMetricsStore = ruleRunMetricsStore;
this.apiKey = apiKey;
this.ruleConsumer = ruleConsumer;
this.executionId = executionId;
this.ruleLabel = ruleLabel;
this.actionsClient = actionsClient;
this.ephemeralActionsToSchedule = taskRunnerContext.maxEphemeralActionsPerRule;
this.ruleTypeActionGroups = new Map(
ruleType.actionGroups.map((actionGroup) => [actionGroup.id, actionGroup.name])
);
this.mutedAlertIdsSet = new Set(rule.mutedInstanceIds);
}
public async run(
alerts: Record<string, Alert<State, Context, ActionGroupIds | RecoveryActionGroupId>>,
recovered: boolean = false
) {
const {
CHUNK_SIZE,
logger,
alertingEventLogger,
ruleRunMetricsStore,
taskRunnerContext: { actionsConfigMap, actionsPlugin },
taskInstance: {
params: { spaceId, alertId: ruleId },
},
} = this;
const executables = this.generateExecutables({ alerts, recovered });
if (!!executables.length) {
const logActions = [];
const bulkActions: EnqueueExecutionOptions[] = [];
this.ruleRunMetricsStore.incrementNumberOfGeneratedActions(executables.length);
for (const { action, alert, alertId, actionGroup, state } of executables) {
const { actionTypeId } = action;
if (!recovered) {
alert.updateLastScheduledActions(action.group as ActionGroupIds);
alert.unscheduleActions();
}
ruleRunMetricsStore.incrementNumberOfGeneratedActionsByConnectorType(actionTypeId);
if (ruleRunMetricsStore.hasReachedTheExecutableActionsLimit(actionsConfigMap)) {
ruleRunMetricsStore.setTriggeredActionsStatusByConnectorType({
actionTypeId,
status: ActionsCompletion.PARTIAL,
});
logger.debug(
`Rule "${this.rule.id}" skipped scheduling action "${action.id}" because the maximum number of allowed actions has been reached.`
);
break;
}
if (
ruleRunMetricsStore.hasReachedTheExecutableActionsLimitByConnectorType({
actionTypeId,
actionsConfigMap,
})
) {
if (!ruleRunMetricsStore.hasConnectorTypeReachedTheLimit(actionTypeId)) {
logger.debug(
`Rule "${this.rule.id}" skipped scheduling action "${action.id}" because the maximum number of allowed actions for connector type ${actionTypeId} has been reached.`
);
}
ruleRunMetricsStore.setTriggeredActionsStatusByConnectorType({
actionTypeId,
status: ActionsCompletion.PARTIAL,
});
continue;
}
if (!this.isActionExecutable(action)) {
this.logger.warn(
`Rule "${this.taskInstance.params.alertId}" skipped scheduling action "${action.id}" because it is disabled`
);
continue;
}
ruleRunMetricsStore.incrementNumberOfTriggeredActions();
ruleRunMetricsStore.incrementNumberOfTriggeredActionsByConnectorType(actionTypeId);
const actionToRun = {
...action,
params: injectActionParams({
ruleId,
spaceId,
actionTypeId,
actionParams: transformActionParams({
actionsPlugin,
alertId: ruleId,
alertType: this.ruleType.id,
actionTypeId,
alertName: this.rule.name,
spaceId,
tags: this.rule.tags,
alertInstanceId: alertId,
alertActionGroup: actionGroup,
alertActionGroupName: this.ruleTypeActionGroups!.get(actionGroup)!,
context: alert.getContext(),
actionId: action.id,
state,
kibanaBaseUrl: this.taskRunnerContext.kibanaBaseUrl,
alertParams: this.rule.params,
actionParams: action.params,
}),
}),
};
await this.actionRunOrAddToBulk({
enqueueOptions: this.getEnqueueOptions(actionToRun),
bulkActions,
});
logActions.push({
id: action.id,
typeId: action.actionTypeId,
alertId,
alertGroup: action.group,
});
if (recovered) {
alert.scheduleActions(action.group as ActionGroupIds);
}
}
if (!!bulkActions.length) {
for (const c of chunk(bulkActions, CHUNK_SIZE)) {
await this.actionsClient!.bulkEnqueueExecution(c);
}
}
if (!!logActions.length) {
for (const action of logActions) {
alertingEventLogger.logAction(action);
}
}
}
}
private generateExecutables({
alerts,
recovered,
}: {
alerts: Record<string, Alert<State, Context, ActionGroupIds | RecoveryActionGroupId>>;
recovered: boolean;
}) {
const executables = [];
for (const action of this.rule.actions) {
for (const [alertId, alert] of Object.entries(alerts)) {
const actionGroup = recovered
? this.ruleType.recoveryActionGroup.id
: alert.getScheduledActionOptions()?.actionGroup!;
if (!this.ruleTypeActionGroups!.has(actionGroup)) {
this.logger.error(
`Invalid action group "${actionGroup}" for rule "${this.ruleType.id}".`
);
continue;
}
if (action.group === actionGroup && this.isAlertExecutable({ alertId, alert, recovered })) {
const state = recovered ? {} : alert.getScheduledActionOptions()?.state!;
executables.push({
action,
alert,
alertId,
actionGroup,
state,
});
}
}
}
return executables;
}
private async actionRunOrAddToBulk({
enqueueOptions,
bulkActions,
}: {
enqueueOptions: EnqueueExecutionOptions;
bulkActions: EnqueueExecutionOptions[];
}) {
if (this.taskRunnerContext.supportsEphemeralTasks && this.ephemeralActionsToSchedule > 0) {
this.ephemeralActionsToSchedule--;
try {
await this.actionsClient!.ephemeralEnqueuedExecution(enqueueOptions);
} catch (err) {
if (isEphemeralTaskRejectedDueToCapacityError(err)) {
bulkActions.push(enqueueOptions);
}
}
} else {
bulkActions.push(enqueueOptions);
}
}
private getEnqueueOptions(action: RuleAction): EnqueueExecutionOptions {
const {
apiKey,
ruleConsumer,
executionId,
taskInstance: {
params: { spaceId, alertId: ruleId },
},
} = this;
const namespace = spaceId === 'default' ? {} : { namespace: spaceId };
return {
id: action.id,
params: action.params,
spaceId,
apiKey: apiKey ?? null,
consumer: ruleConsumer,
source: asSavedObjectExecutionSource({
id: ruleId,
type: 'alert',
}),
executionId,
relatedSavedObjects: [
{
id: ruleId,
type: 'alert',
namespace: namespace.namespace,
typeId: this.ruleType.id,
},
],
};
}
private isActionExecutable(action: RuleAction) {
return this.taskRunnerContext.actionsPlugin.isActionExecutable(action.id, action.actionTypeId, {
notifyUsage: true,
});
}
private isAlertExecutable({
alertId,
alert,
recovered,
}: {
alertId: string;
alert: Alert<AlertInstanceState, AlertInstanceContext, ActionGroupIds | RecoveryActionGroupId>;
recovered: boolean;
}) {
const {
rule: { throttle, notifyWhen },
ruleLabel,
logger,
mutedAlertIdsSet,
} = this;
const muted = mutedAlertIdsSet!.has(alertId);
const throttled = alert.isThrottled(throttle);
if (muted) {
if (
!this.skippedAlerts[alertId] ||
(this.skippedAlerts[alertId] && this.skippedAlerts[alertId].reason !== Reasons.MUTED)
) {
logger.debug(
`skipping scheduling of actions for '${alertId}' in rule ${ruleLabel}: rule is muted`
);
}
this.skippedAlerts[alertId] = { reason: Reasons.MUTED };
return false;
}
if (!recovered) {
if (throttled) {
if (
!this.skippedAlerts[alertId] ||
(this.skippedAlerts[alertId] && this.skippedAlerts[alertId].reason !== Reasons.THROTTLED)
) {
logger.debug(
`skipping scheduling of actions for '${alertId}' in rule ${ruleLabel}: rule is throttled`
);
}
this.skippedAlerts[alertId] = { reason: Reasons.THROTTLED };
return false;
}
if (notifyWhen === 'onActionGroupChange' && !alert.scheduledActionGroupHasChanged()) {
if (
!this.skippedAlerts[alertId] ||
(this.skippedAlerts[alertId] &&
this.skippedAlerts[alertId].reason !== Reasons.ACTION_GROUP_NOT_CHANGED)
) {
logger.debug(
`skipping scheduling of actions for '${alertId}' in rule ${ruleLabel}: alert is active but action group has not changed`
);
}
this.skippedAlerts[alertId] = { reason: Reasons.ACTION_GROUP_NOT_CHANGED };
return false;
}
return alert.hasScheduledActions();
} else {
return true;
}
}
}

207
x-pack/plugins/alerting/server/task_runner/schedule_actions_for_alerts.test.ts
View file

@ -1,207 +0,0 @@
/*
* 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 { RuleRunMetricsStore } from '../lib/rule_run_metrics_store';
import { RecoveredActionGroup } from '../types';
import { RULE_NAME } from './fixtures';
import { loggingSystemMock } from '@kbn/core/server/mocks';
import { scheduleActionsForAlerts } from './schedule_actions_for_alerts';
import { Alert } from '../alert';
import { AlertInstanceState, AlertInstanceContext, DefaultActionGroupId } from '../../common';
import sinon from 'sinon';
describe('Schedule Actions For Alerts', () => {
const ruleRunMetricsStore = new RuleRunMetricsStore();
const executionHandler = jest.fn();
const recoveryActionGroup = RecoveredActionGroup;
const mutedAlertIdsSet = new Set('2');
const logger: ReturnType<typeof loggingSystemMock.createLogger> =
loggingSystemMock.createLogger();
const notifyWhen = 'onActiveAlert';
const throttle = null;
let clock: sinon.SinonFakeTimers;
beforeEach(() => {
jest.resetAllMocks();
clock.reset();
});
beforeAll(() => {
clock = sinon.useFakeTimers();
});
afterAll(() => clock.restore());
test('schedules alerts with executable actions', async () => {
const alert = new Alert<AlertInstanceState, AlertInstanceContext, DefaultActionGroupId>('1', {
state: { test: true },
meta: {
lastScheduledActions: {
date: new Date(),
group: 'default',
},
},
});
alert.scheduleActions('default');
const alerts = { '1': alert };
const recoveredAlerts = {};
await scheduleActionsForAlerts({
activeAlerts: alerts,
recoveryActionGroup,
recoveredAlerts,
executionHandler,
mutedAlertIdsSet,
logger,
ruleLabel: RULE_NAME,
ruleRunMetricsStore,
throttle,
notifyWhen,
});
expect(executionHandler).toBeCalledWith({
actionGroup: 'default',
context: {},
state: { test: true },
alertId: '1',
ruleRunMetricsStore,
});
});
test('schedules alerts with recovered actions', async () => {
const alert = new Alert<AlertInstanceState, AlertInstanceContext, DefaultActionGroupId>('1', {
state: { test: true },
meta: {
lastScheduledActions: {
date: new Date(),
group: 'default',
},
},
});
const alerts = {};
const recoveredAlerts = { '1': alert };
await scheduleActionsForAlerts({
activeAlerts: alerts,
recoveryActionGroup,
recoveredAlerts,
executionHandler,
mutedAlertIdsSet,
logger,
ruleLabel: RULE_NAME,
ruleRunMetricsStore,
throttle,
notifyWhen,
});
expect(executionHandler).toHaveBeenNthCalledWith(1, {
actionGroup: 'recovered',
context: {},
state: {},
alertId: '1',
ruleRunMetricsStore,
});
});
test('does not schedule alerts with recovered actions that are muted', async () => {
const alert = new Alert<AlertInstanceState, AlertInstanceContext, DefaultActionGroupId>('2', {
state: { test: true },
meta: {
lastScheduledActions: {
date: new Date(),
group: 'default',
},
},
});
const alerts = {};
const recoveredAlerts = { '2': alert };
await scheduleActionsForAlerts({
activeAlerts: alerts,
recoveryActionGroup,
recoveredAlerts,
executionHandler,
mutedAlertIdsSet,
logger,
ruleLabel: RULE_NAME,
ruleRunMetricsStore,
throttle,
notifyWhen,
});
expect(executionHandler).not.toBeCalled();
expect(logger.debug).nthCalledWith(
1,
`skipping scheduling of actions for '2' in rule ${RULE_NAME}: instance is muted`
);
});
test('does not schedule active alerts that are throttled', async () => {
const alert = new Alert<AlertInstanceState, AlertInstanceContext, DefaultActionGroupId>('1', {
state: { test: true },
meta: {
lastScheduledActions: {
date: new Date(),
group: 'default',
},
},
});
clock.tick(30000);
alert.scheduleActions('default');
const alerts = { '1': alert };
const recoveredAlerts = {};
await scheduleActionsForAlerts({
activeAlerts: alerts,
recoveryActionGroup,
recoveredAlerts,
executionHandler,
mutedAlertIdsSet,
logger,
ruleLabel: RULE_NAME,
ruleRunMetricsStore,
throttle: '1m',
notifyWhen,
});
expect(executionHandler).not.toBeCalled();
expect(logger.debug).nthCalledWith(
1,
`skipping scheduling of actions for '1' in rule ${RULE_NAME}: rule is throttled`
);
});
test('does not schedule active alerts that are muted', async () => {
const alert = new Alert<AlertInstanceState, AlertInstanceContext, DefaultActionGroupId>('2', {
state: { test: true },
meta: {
lastScheduledActions: {
date: new Date(),
group: 'default',
},
},
});
const alerts = { '2': alert };
const recoveredAlerts = {};
await scheduleActionsForAlerts({
activeAlerts: alerts,
recoveryActionGroup,
recoveredAlerts,
executionHandler,
mutedAlertIdsSet,
logger,
ruleLabel: RULE_NAME,
ruleRunMetricsStore,
throttle,
notifyWhen,
});
expect(executionHandler).not.toBeCalled();
expect(logger.debug).nthCalledWith(
1,
`skipping scheduling of actions for '2' in rule ${RULE_NAME}: rule is muted`
);
});
});

134
x-pack/plugins/alerting/server/task_runner/schedule_actions_for_alerts.ts
View file

@ -1,134 +0,0 @@
/*
* 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 { Logger } from '@kbn/core/server';
import { ExecutionHandler } from './create_execution_handler';
import { ScheduleActionsForAlertsParams } from './types';
import { AlertInstanceState, AlertInstanceContext } from '../types';
import { Alert } from '../alert';
import { RuleRunMetricsStore } from '../lib/rule_run_metrics_store';
export async function scheduleActionsForAlerts<
InstanceState extends AlertInstanceState,
InstanceContext extends AlertInstanceContext,
ActionGroupIds extends string,
RecoveryActionGroupId extends string
>(
params: ScheduleActionsForAlertsParams<
InstanceState,
InstanceContext,
ActionGroupIds,
RecoveryActionGroupId
>
): Promise<void> {
const {
logger,
activeAlerts,
recoveryActionGroup,
recoveredAlerts,
executionHandler,
mutedAlertIdsSet,
ruleLabel,
ruleRunMetricsStore,
throttle,
notifyWhen,
} = params;
// execute alerts with executable actions
for (const [alertId, alert] of Object.entries(activeAlerts)) {
const executeAction: boolean = shouldExecuteAction(
alertId,
alert,
mutedAlertIdsSet,
ruleLabel,
logger,
throttle,
notifyWhen
);
if (executeAction && alert.hasScheduledActions()) {
const { actionGroup, state } = alert.getScheduledActionOptions()!;
await executeAlert(alertId, alert, executionHandler, ruleRunMetricsStore, actionGroup, state);
}
}
// execute recovered alerts
for (const alertId of Object.keys(recoveredAlerts)) {
if (mutedAlertIdsSet.has(alertId)) {
logger.debug(
`skipping scheduling of actions for '${alertId}' in rule ${ruleLabel}: instance is muted`
);
} else {
const alert = recoveredAlerts[alertId];
await executeAlert(
alertId,
alert,
executionHandler,
ruleRunMetricsStore,
recoveryActionGroup.id,
{} as InstanceState
);
alert.scheduleActions(recoveryActionGroup.id);
}
}
}
async function executeAlert<
InstanceState extends AlertInstanceState,
InstanceContext extends AlertInstanceContext,
ActionGroupIds extends string,
RecoveryActionGroupId extends string
>(
alertId: string,
alert: Alert<InstanceState, InstanceContext, ActionGroupIds | RecoveryActionGroupId>,
executionHandler: ExecutionHandler<ActionGroupIds | RecoveryActionGroupId>,
ruleRunMetricsStore: RuleRunMetricsStore,
actionGroup: ActionGroupIds | RecoveryActionGroupId,
state: InstanceState
) {
alert.updateLastScheduledActions(actionGroup);
alert.unscheduleActions();
return executionHandler({
actionGroup,
context: alert.getContext(),
state,
alertId,
ruleRunMetricsStore,
});
}
function shouldExecuteAction<
InstanceState extends AlertInstanceState,
InstanceContext extends AlertInstanceContext,
ActionGroupIds extends string
>(
alertId: string,
alert: Alert<InstanceState, InstanceContext, ActionGroupIds>,
mutedAlertIdsSet: Set<string>,
ruleLabel: string,
logger: Logger,
throttle: string | null,
notifyWhen: string | null
) {
const throttled = alert.isThrottled(throttle);
const muted = mutedAlertIdsSet.has(alertId);
let executeAction = true;
if (throttled || muted) {
executeAction = false;
logger.debug(
`skipping scheduling of actions for '${alertId}' in rule ${ruleLabel}: rule is ${
muted ? 'muted' : 'throttled'
}`
);
} else if (notifyWhen === 'onActionGroupChange' && !alert.scheduledActionGroupHasChanged()) {
executeAction = false;
logger.debug(
`skipping scheduling of actions for '${alertId}' in rule ${ruleLabel}: alert is active but action group has not changed`
);
}
return executeAction;
}

5
x-pack/plugins/alerting/server/task_runner/task_runner.test.ts
View file

@ -2424,8 +2424,7 @@ describe('Task Runner', () => {
const runnerResult = await taskRunner.run();
// 1x(.server-log) and 1x(any-action) per alert
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(2);
expect(actionsClient.bulkEnqueueExecution).toHaveBeenCalledTimes(1);
expect(
taskRunnerFactoryInitializerParams.internalSavedObjectsRepository.update
@ -2468,7 +2467,7 @@ describe('Task Runner', () => {
expect(logger.debug).nthCalledWith(
3,
'Rule "1" skipped scheduling action "2" because the maximum number of allowed actions for connector type .server-log has been reached.'
'Rule "1" skipped scheduling action "1" because the maximum number of allowed actions for connector type .server-log has been reached.'
);
testAlertingEventLogCalls({

120
x-pack/plugins/alerting/server/task_runner/task_runner.ts
View file

@ -9,11 +9,11 @@ import apm from 'elastic-apm-node';
import { cloneDeep, omit } from 'lodash';
import { UsageCounter } from '@kbn/usage-collection-plugin/server';
import uuid from 'uuid';
import { KibanaRequest, Logger } from '@kbn/core/server';
import { Logger } from '@kbn/core/server';
import { ConcreteTaskInstance, throwUnrecoverableError } from '@kbn/task-manager-plugin/server';
import { nanosToMillis } from '@kbn/event-log-plugin/server';
import { ExecutionHandler } from './execution_handler';
import { TaskRunnerContext } from './task_runner_factory';
import { createExecutionHandler } from './create_execution_handler';
import { Alert, createAlertFactory } from '../alert';
import {
ElasticsearchError,
@ -25,12 +25,10 @@ import {
processAlerts,
} from '../lib';
import {
Rule,
RuleExecutionStatus,
RuleExecutionStatusErrorReasons,
IntervalSchedule,
RawAlertInstance,
RawRule,
RawRuleExecutionStatus,
RuleMonitoring,
RuleMonitoringHistory,
@ -66,7 +64,6 @@ import { wrapSearchSourceClient } from '../lib/wrap_search_source_client';
import { AlertingEventLogger } from '../lib/alerting_event_logger/alerting_event_logger';
import { loadRule } from './rule_loader';
import { logAlerts } from './log_alerts';
import { scheduleActionsForAlerts } from './schedule_actions_for_alerts';
import { getPublicAlertFactory } from '../alert/create_alert_factory';
import { TaskRunnerTimer, TaskRunnerTimerSpan } from './task_runner_timer';
@ -154,47 +151,6 @@ export class TaskRunner<
this.stackTraceLog = null;
}
private getExecutionHandler(
ruleId: string,
ruleName: string,
tags: string[] | undefined,
spaceId: string,
apiKey: RawRule['apiKey'],
kibanaBaseUrl: string | undefined,
actions: Rule<Params>['actions'],
ruleParams: Params,
request: KibanaRequest
) {
return createExecutionHandler<
Params,
ExtractedParams,
RuleState,
State,
Context,
ActionGroupIds,
RecoveryActionGroupId
>({
ruleId,
ruleName,
ruleConsumer: this.ruleConsumer!,
tags,
executionId: this.executionId,
logger: this.logger,
actionsPlugin: this.context.actionsPlugin,
apiKey,
actions,
spaceId,
ruleType: this.ruleType,
kibanaBaseUrl,
alertingEventLogger: this.alertingEventLogger,
request,
ruleParams,
supportsEphemeralTasks: this.context.supportsEphemeralTasks,
maxEphemeralActionsPerRule: this.context.maxEphemeralActionsPerRule,
actionsConfigMap: this.context.actionsConfigMap,
});
}
private async updateRuleSavedObject(
ruleId: string,
namespace: string | undefined,
@ -223,6 +179,11 @@ export class TaskRunner<
return !this.context.cancelAlertsOnRuleTimeout || !this.ruleType.cancelAlertsOnRuleTimeout;
}
// Usage counter for telemetry
// This keeps track of how many times action executions were skipped after rule
// execution completed successfully after the execution timeout
// This can occur when rule executors do not short circuit execution in response
// to timeout
private countUsageOfActionExecutionAfterRuleCancellation() {
if (this.cancelled && this.usageCounter) {
if (this.context.cancelAlertsOnRuleTimeout && this.ruleType.cancelAlertsOnRuleTimeout) {
@ -259,7 +220,6 @@ export class TaskRunner<
schedule,
throttle,
notifyWhen,
mutedInstanceIds,
name,
tags,
createdBy,
@ -464,52 +424,34 @@ export class TaskRunner<
}
);
await this.timer.runWithTimer(TaskRunnerTimerSpan.TriggerActions, async () => {
const executionHandler = this.getExecutionHandler(
ruleId,
rule.name,
rule.tags,
spaceId,
apiKey,
this.context.kibanaBaseUrl,
rule.actions,
rule.params,
fakeRequest
);
const executionHandler = new ExecutionHandler({
rule,
ruleType: this.ruleType,
logger: this.logger,
taskRunnerContext: this.context,
taskInstance: this.taskInstance,
ruleRunMetricsStore,
apiKey,
ruleConsumer: this.ruleConsumer!,
executionId: this.executionId,
ruleLabel,
alertingEventLogger: this.alertingEventLogger,
actionsClient: await this.context.actionsPlugin.getActionsClientWithRequest(fakeRequest),
});
await this.timer.runWithTimer(TaskRunnerTimerSpan.TriggerActions, async () => {
await rulesClient.clearExpiredSnoozes({ id: rule.id });
const ruleIsSnoozed = isRuleSnoozed(rule);
if (!ruleIsSnoozed && this.shouldLogAndScheduleActionsForAlerts()) {
const mutedAlertIdsSet = new Set(mutedInstanceIds);
await scheduleActionsForAlerts<State, Context, ActionGroupIds, RecoveryActionGroupId>({
activeAlerts,
recoveryActionGroup: this.ruleType.recoveryActionGroup,
recoveredAlerts,
executionHandler,
mutedAlertIdsSet,
logger: this.logger,
ruleLabel,
ruleRunMetricsStore,
throttle,
notifyWhen,
});
if (isRuleSnoozed(rule)) {
this.logger.debug(`no scheduling of actions for rule ${ruleLabel}: rule is snoozed.`);
} else if (!this.shouldLogAndScheduleActionsForAlerts()) {
this.logger.debug(
`no scheduling of actions for rule ${ruleLabel}: rule execution has been cancelled.`
);
this.countUsageOfActionExecutionAfterRuleCancellation();
} else {
if (ruleIsSnoozed) {
this.logger.debug(`no scheduling of actions for rule ${ruleLabel}: rule is snoozed.`);
}
if (!this.shouldLogAndScheduleActionsForAlerts()) {
this.logger.debug(
`no scheduling of actions for rule ${ruleLabel}: rule execution has been cancelled.`
);
// Usage counter for telemetry
// This keeps track of how many times action executions were skipped after rule
// execution completed successfully after the execution timeout
// This can occur when rule executors do not short circuit execution in response
// to timeout
this.countUsageOfActionExecutionAfterRuleCancellation();
}
await executionHandler.run(activeAlerts);
await executionHandler.run(recoveredAlerts, true);
}
});

70
x-pack/plugins/alerting/server/task_runner/types.ts
View file

@ -7,25 +7,21 @@
import { KibanaRequest, Logger } from '@kbn/core/server';
import { ConcreteTaskInstance } from '@kbn/task-manager-plugin/server';
import { PluginStartContract as ActionsPluginStartContract } from '@kbn/actions-plugin/server';
import { PublicMethodsOf } from '@kbn/utility-types';
import { ActionsClient } from '@kbn/actions-plugin/server/actions_client';
import { TaskRunnerContext } from './task_runner_factory';
import {
ActionGroup,
RuleAction,
AlertInstanceContext,
AlertInstanceState,
RuleTypeParams,
RuleTypeState,
IntervalSchedule,
RuleMonitoring,
RuleTaskState,
SanitizedRule,
RuleTypeState,
} from '../../common';
import { Alert } from '../alert';
import { NormalizedRuleType } from '../rule_type_registry';
import { ExecutionHandler } from './create_execution_handler';
import { RawRule, RulesClientApi } from '../types';
import { ActionsConfigMap } from '../lib/get_actions_config_map';
import { RuleRunMetrics, RuleRunMetricsStore } from '../lib/rule_run_metrics_store';
import { AlertingEventLogger } from '../lib/alerting_event_logger/alerting_event_logger';
@ -57,67 +53,35 @@ export interface RuleTaskInstance extends ConcreteTaskInstance {
state: RuleTaskState;
}
export interface ScheduleActionsForAlertsParams<
InstanceState extends AlertInstanceState,
InstanceContext extends AlertInstanceContext,
ActionGroupIds extends string,
RecoveryActionGroupId extends string
> {
logger: Logger;
recoveryActionGroup: ActionGroup<RecoveryActionGroupId>;
recoveredAlerts: Record<string, Alert<InstanceState, InstanceContext, RecoveryActionGroupId>>;
executionHandler: ExecutionHandler<ActionGroupIds | RecoveryActionGroupId>;
mutedAlertIdsSet: Set<string>;
ruleLabel: string;
ruleRunMetricsStore: RuleRunMetricsStore;
activeAlerts: Record<string, Alert<InstanceState, InstanceContext, ActionGroupIds>>;
throttle: string | null;
notifyWhen: string | null;
}
// / ExecutionHandler
export interface CreateExecutionHandlerOptions<
export interface ExecutionHandlerOptions<
Params extends RuleTypeParams,
ExtractedParams extends RuleTypeParams,
State extends RuleTypeState,
InstanceState extends AlertInstanceState,
InstanceContext extends AlertInstanceContext,
RuleState extends RuleTypeState,
State extends AlertInstanceState,
Context extends AlertInstanceContext,
ActionGroupIds extends string,
RecoveryActionGroupId extends string
> {
ruleId: string;
ruleName: string;
ruleConsumer: string;
executionId: string;
tags?: string[];
actionsPlugin: ActionsPluginStartContract;
actions: RuleAction[];
spaceId: string;
apiKey: RawRule['apiKey'];
kibanaBaseUrl: string | undefined;
ruleType: NormalizedRuleType<
Params,
ExtractedParams,
RuleState,
State,
InstanceState,
InstanceContext,
Context,
ActionGroupIds,
RecoveryActionGroupId
>;
logger: Logger;
alertingEventLogger: PublicMethodsOf<AlertingEventLogger>;
request: KibanaRequest;
ruleParams: RuleTypeParams;
supportsEphemeralTasks: boolean;
maxEphemeralActionsPerRule: number;
actionsConfigMap: ActionsConfigMap;
}
export interface ExecutionHandlerOptions<ActionGroupIds extends string> {
actionGroup: ActionGroupIds;
alertId: string;
context: AlertInstanceContext;
state: AlertInstanceState;
rule: SanitizedRule<Params>;
taskRunnerContext: TaskRunnerContext;
taskInstance: RuleTaskInstance;
ruleRunMetricsStore: RuleRunMetricsStore;
apiKey: RawRule['apiKey'];
ruleConsumer: string;
executionId: string;
ruleLabel: string;
actionsClient: PublicMethodsOf<ActionsClient>;
}