Skip to content

[Task Manager] Adding list of explicitly de-registered task types #123963

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ymao1 merged 16 commits into from
Feb 15, 2022
Merged

[Task Manager] Adding list of explicitly de-registered task types #123963

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
1ef085d
Adding REMOVED_TYPES to task manager and only marking those types as …
ymao1 Jan 27, 2022
cf0726d
Merge branch 'main' of https://github.com/elastic/kibana into task-ma…
ymao1 Jan 28, 2022
7dbccbc
Adding unit tests
ymao1 Jan 28, 2022
976a14f
Fixing functional test
ymao1 Jan 28, 2022
1a0453b
Merge branch 'main' of https://github.com/elastic/kibana into task-ma…
ymao1 Jan 31, 2022
7a5cf7b
Throwing error when registering a removed task type
ymao1 Jan 31, 2022
e126374
Adding migration
ymao1 Jan 31, 2022
821bc7d
Adding functional tests
ymao1 Jan 31, 2022
ed31d92
Merge branch 'main' into task-manager/unregistered-task-types
kibanamachine Feb 2, 2022
0c40a92
Merge branch 'main' into task-manager/unregistered-task-types
kibanamachine Feb 3, 2022
249a912
Merging in main
ymao1 Feb 9, 2022
770cb4a
Merge branch 'task-manager/unregistered-task-types' of https://github…
ymao1 Feb 9, 2022
bda2fdd
Cleanup
ymao1 Feb 9, 2022
09d30cd
Merge branch 'main' into task-manager/unregistered-task-types
kibanamachine Feb 14, 2022
2fa9e16
Merge branch 'main' into task-manager/unregistered-task-types
kibanamachine Feb 15, 2022
2dee057
Adding disabled siem signals rule type
ymao1 Feb 15, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion x-pack/plugins/task_manager/server/plugin.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ import { TaskManagerConfig } from './config';
import { createInitialMiddleware, addMiddlewareToChain, Middleware } from './lib/middleware';
import { removeIfExists } from './lib/remove_if_exists';
import { setupSavedObjects } from './saved_objects';
import { TaskDefinitionRegistry, TaskTypeDictionary } from './task_type_dictionary';
import { TaskDefinitionRegistry, TaskTypeDictionary, REMOVED_TYPES } from './task_type_dictionary';
import { FetchResult, SearchOpts, TaskStore } from './task_store';
import { createManagedConfiguration } from './lib/create_managed_configuration';
import { TaskScheduling } from './task_scheduling';
Expand Down Expand Up @@ -189,6 +189,7 @@ export class TaskManagerPlugin
this.taskPollingLifecycle = new TaskPollingLifecycle({
config: this.config!,
definitions: this.definitions,
unusedTypes: REMOVED_TYPES,
logger: this.logger,
executionContext,
taskStore,
Expand Down
1 change: 1 addition & 0 deletions x-pack/plugins/task_manager/server/polling_lifecycle.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,7 @@ describe('TaskPollingLifecycle', () => {
},
taskStore: mockTaskStore,
logger: taskManagerLogger,
unusedTypes: [],
definitions: new TaskTypeDictionary(taskManagerLogger),
middleware: createInitialMiddleware(),
maxWorkersConfiguration$: of(100),
Expand Down
3 changes: 3 additions & 0 deletions x-pack/plugins/task_manager/server/polling_lifecycle.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@ import { TaskClaiming, ClaimOwnershipResult } from './queries/task_claiming';
export type TaskPollingLifecycleOpts = {
logger: Logger;
definitions: TaskTypeDictionary;
unusedTypes: string[];
taskStore: TaskStore;
config: TaskManagerConfig;
middleware: Middleware;
Expand Down Expand Up @@ -106,6 +107,7 @@ export class TaskPollingLifecycle {
config,
taskStore,
definitions,
unusedTypes,
executionContext,
usageCounter,
}: TaskPollingLifecycleOpts) {
Expand Down Expand Up @@ -134,6 +136,7 @@ export class TaskPollingLifecycle {
maxAttempts: config.max_attempts,
excludedTaskTypes: config.unsafe.exclude_task_types,
definitions,
unusedTypes,
logger: this.logger,
getCapacity: (taskType?: string) =>
taskType && this.definitions.get(taskType)?.maxConcurrency
Expand Down
47 changes: 32 additions & 15 deletions x-pack/plugins/task_manager/server/queries/mark_available_tasks_as_claimed.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,15 +47,16 @@ describe('mark_available_tasks_as_claimed', () => {
// status running or claiming with a retryAt <= now.
shouldBeOneOf(IdleTaskWithExpiredRunAt, RunningOrClaimingTaskWithExpiredRetryAt)
),
script: updateFieldsAndMarkAsFailed(
script: updateFieldsAndMarkAsFailed({
fieldUpdates,
claimTasksById || [],
definitions.getAllTypes(),
[],
Array.from(definitions).reduce((accumulator, [type, { maxAttempts }]) => {
claimTasksById: claimTasksById || [],
claimableTaskTypes: definitions.getAllTypes(),
skippedTaskTypes: [],
unusedTaskTypes: [],
taskMaxAttempts: Array.from(definitions).reduce((accumulator, [type, { maxAttempts }]) => {
return { ...accumulator, [type]: maxAttempts || defaultMaxAttempts };
}, {})
),
}, {}),
}),
sort: SortByRunAtAndRetryAt,
}).toEqual({
query: {
Expand Down Expand Up @@ -126,7 +127,7 @@ if (doc['task.runAt'].size()!=0) {
ctx._source.task.status = "claiming"; ${Object.keys(fieldUpdates)
.map((field) => `ctx._source.task.${field}=params.fieldUpdates.${field};`)
.join(' ')}
} else if (!params.skippedTaskTypes.contains(ctx._source.task.taskType)) {
Copy link
Copy Markdown
Contributor

@ersin-erdal ersin-erdal Feb 10, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Don't we need to test the skipped tasks?

Copy link
Copy Markdown
Contributor Author

@ymao1 ymao1 Feb 10, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Previously we maintained a list of claimableTaskTypes and skippedTaskTypes. These are all the types that this task manager knows about so if a task is not claimable or skipped, we mark is as unrecognized. Now we are just using claimableTaskTypes to claim and unusedTaskTypes to determine what to mark as unrecognized. Skipped task types will go into the else bucket and get a no-op.

} else if (params.unusedTaskTypes.contains(ctx._source.task.taskType)) {
ctx._source.task.status = "unrecognized";
} else {
ctx.op = "noop";
Expand All @@ -140,6 +141,7 @@ if (doc['task.runAt'].size()!=0) {
claimTasksById: [],
claimableTaskTypes: ['sampleTask', 'otherTask'],
skippedTaskTypes: [],
unusedTaskTypes: [],
taskMaxAttempts: {
sampleTask: 5,
otherTask: 1,
Expand All @@ -164,9 +166,16 @@ if (doc['task.runAt'].size()!=0) {
];

expect(
updateFieldsAndMarkAsFailed(fieldUpdates, claimTasksById, ['foo', 'bar'], [], {
foo: 5,
bar: 2,
updateFieldsAndMarkAsFailed({
Copy link
Copy Markdown
Contributor

@ersin-erdal ersin-erdal Feb 15, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Clean Code, group the arguments, love it :)

fieldUpdates,
claimTasksById,
claimableTaskTypes: ['foo', 'bar'],
skippedTaskTypes: [],
unusedTaskTypes: [],
taskMaxAttempts: {
foo: 5,
bar: 2,
},
})
).toMatchObject({
source: `
Expand All @@ -182,7 +191,7 @@ if (doc['task.runAt'].size()!=0) {
ctx._source.task.status = "claiming"; ${Object.keys(fieldUpdates)
.map((field) => `ctx._source.task.${field}=params.fieldUpdates.${field};`)
.join(' ')}
} else if (!params.skippedTaskTypes.contains(ctx._source.task.taskType)) {
} else if (params.unusedTaskTypes.contains(ctx._source.task.taskType)) {
ctx._source.task.status = "unrecognized";
} else {
ctx.op = "noop";
Expand All @@ -196,6 +205,7 @@ if (doc['task.runAt'].size()!=0) {
],
claimableTaskTypes: ['foo', 'bar'],
skippedTaskTypes: [],
unusedTaskTypes: [],
taskMaxAttempts: {
foo: 5,
bar: 2,
Expand All @@ -213,9 +223,16 @@ if (doc['task.runAt'].size()!=0) {
};

expect(
updateFieldsAndMarkAsFailed(fieldUpdates, [], ['foo', 'bar'], [], {
foo: 5,
bar: 2,
updateFieldsAndMarkAsFailed({
fieldUpdates,
claimTasksById: [],
claimableTaskTypes: ['foo', 'bar'],
skippedTaskTypes: [],
unusedTaskTypes: [],
taskMaxAttempts: {
foo: 5,
bar: 2,
},
}).source
).toMatch(/ctx.op = "noop"/);
});
Expand Down
27 changes: 19 additions & 8 deletions x-pack/plugins/task_manager/server/queries/mark_available_tasks_as_claimed.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,15 +104,25 @@ if (doc['task.runAt'].size()!=0) {
};
export const SortByRunAtAndRetryAt = SortByRunAtAndRetryAtScript as estypes.SortCombinations;

export const updateFieldsAndMarkAsFailed = (
export interface UpdateFieldsAndMarkAsFailedOpts {
fieldUpdates: {
[field: string]: string | number | Date;
},
claimTasksById: string[],
claimableTaskTypes: string[],
skippedTaskTypes: string[],
taskMaxAttempts: { [field: string]: number }
): ScriptClause => {
};
claimTasksById: string[];
claimableTaskTypes: string[];
skippedTaskTypes: string[];
unusedTaskTypes: string[];
taskMaxAttempts: { [field: string]: number };
}

export const updateFieldsAndMarkAsFailed = ({
fieldUpdates,
claimTasksById,
claimableTaskTypes,
skippedTaskTypes,
unusedTaskTypes,
taskMaxAttempts,
}: UpdateFieldsAndMarkAsFailedOpts): ScriptClause => {
const markAsClaimingScript = `ctx._source.task.status = "claiming"; ${Object.keys(fieldUpdates)
.map((field) => `ctx._source.task.${field}=params.fieldUpdates.${field};`)
.join(' ')}`;
Expand All @@ -126,7 +136,7 @@ export const updateFieldsAndMarkAsFailed = (
}
} else if (params.skippedTaskTypes.contains(ctx._source.task.taskType) && params.claimTasksById.contains(ctx._id)) {
${markAsClaimingScript}
} else if (!params.skippedTaskTypes.contains(ctx._source.task.taskType)) {
} else if (params.unusedTaskTypes.contains(ctx._source.task.taskType)) {
ctx._source.task.status = "unrecognized";
} else {
ctx.op = "noop";
Expand All @@ -137,6 +147,7 @@ export const updateFieldsAndMarkAsFailed = (
claimTasksById,
claimableTaskTypes,
skippedTaskTypes,
unusedTaskTypes,
taskMaxAttempts,
},
};
Expand Down
125 changes: 125 additions & 0 deletions x-pack/plugins/task_manager/server/queries/task_claiming.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,7 @@ describe('TaskClaiming', () => {
logger: taskManagerLogger,
definitions,
excludedTaskTypes: [],
unusedTypes: [],
taskStore: taskStoreMock.create({ taskManagerId: '' }),
maxAttempts: 2,
getCapacity: () => 10,
Expand All @@ -127,12 +128,14 @@ describe('TaskClaiming', () => {
hits = [generateFakeTasks(1)],
versionConflicts = 2,
excludedTaskTypes = [],
unusedTaskTypes = [],
}: {
storeOpts: Partial<StoreOpts>;
taskClaimingOpts: Partial<TaskClaimingOpts>;
hits?: ConcreteTaskInstance[][];
versionConflicts?: number;
excludedTaskTypes?: string[];
unusedTaskTypes?: string[];
}) {
const definitions = storeOpts.definitions ?? taskDefinitions;
const store = taskStoreMock.create({ taskManagerId: storeOpts.taskManagerId });
Expand Down Expand Up @@ -161,6 +164,7 @@ describe('TaskClaiming', () => {
definitions,
taskStore: store,
excludedTaskTypes,
unusedTypes: unusedTaskTypes,
maxAttempts: taskClaimingOpts.maxAttempts ?? 2,
getCapacity: taskClaimingOpts.getCapacity ?? (() => 10),
...taskClaimingOpts,
Expand All @@ -176,19 +180,22 @@ describe('TaskClaiming', () => {
hits = [generateFakeTasks(1)],
versionConflicts = 2,
excludedTaskTypes = [],
unusedTaskTypes = [],
}: {
storeOpts: Partial<StoreOpts>;
taskClaimingOpts: Partial<TaskClaimingOpts>;
claimingOpts: Omit<OwnershipClaimingOpts, 'size' | 'taskTypes'>;
hits?: ConcreteTaskInstance[][];
versionConflicts?: number;
excludedTaskTypes?: string[];
unusedTaskTypes?: string[];
}) {
const getCapacity = taskClaimingOpts.getCapacity ?? (() => 10);
const { taskClaiming, store } = initialiseTestClaiming({
storeOpts,
taskClaimingOpts,
excludedTaskTypes,
unusedTaskTypes,
hits,
versionConflicts,
});
Expand Down Expand Up @@ -496,6 +503,7 @@ if (doc['task.runAt'].size()!=0) {
],
claimableTaskTypes: ['foo', 'bar'],
skippedTaskTypes: [],
unusedTaskTypes: [],
taskMaxAttempts: {
bar: customMaxAttempts,
foo: maxAttempts,
Expand Down Expand Up @@ -614,6 +622,7 @@ if (doc['task.runAt'].size()!=0) {
'anotherLimitedToOne',
'limitedToTwo',
],
unusedTaskTypes: [],
taskMaxAttempts: {
unlimited: maxAttempts,
},
Expand Down Expand Up @@ -871,6 +880,121 @@ if (doc['task.runAt'].size()!=0) {
expect(firstCycle).not.toMatchObject(secondCycle);
});

test('it passes any unusedTaskTypes to script', async () => {
const maxAttempts = _.random(2, 43);
const customMaxAttempts = _.random(44, 100);
const taskManagerId = uuid.v1();
const fieldUpdates = {
ownerId: taskManagerId,
retryAt: new Date(Date.now()),
};
const definitions = new TaskTypeDictionary(mockLogger());
definitions.registerTaskDefinitions({
foo: {
title: 'foo',
createTaskRunner: jest.fn(),
},
bar: {
title: 'bar',
maxAttempts: customMaxAttempts,
createTaskRunner: jest.fn(),
},
foobar: {
title: 'foobar',
maxAttempts: customMaxAttempts,
createTaskRunner: jest.fn(),
},
});

const [
{
args: {
updateByQuery: [{ query, script }],
},
},
] = await testClaimAvailableTasks({
storeOpts: {
definitions,
taskManagerId,
},
taskClaimingOpts: {
maxAttempts,
},
claimingOpts: {
claimOwnershipUntil: new Date(),
},
excludedTaskTypes: ['foobar'],
unusedTaskTypes: ['barfoo'],
});
expect(query).toMatchObject({
bool: {
must: [
{
bool: {
should: [
{
bool: {
must: [
{ term: { 'task.status': 'idle' } },
{ range: { 'task.runAt': { lte: 'now' } } },
],
},
},
{
bool: {
must: [
{
bool: {
should: [
{ term: { 'task.status': 'running' } },
{ term: { 'task.status': 'claiming' } },
],
},
},
{ range: { 'task.retryAt': { lte: 'now' } } },
],
},
},
],
},
},
],
filter: [
{
bool: {
must_not: [
{
bool: {
should: [
{ term: { 'task.status': 'running' } },
{ term: { 'task.status': 'claiming' } },
],
must: { range: { 'task.retryAt': { gt: 'now' } } },
},
},
],
},
},
],
},
});
expect(script).toMatchObject({
source: expect.any(String),
lang: 'painless',
params: {
fieldUpdates,
claimTasksById: [],
claimableTaskTypes: ['foo', 'bar'],
skippedTaskTypes: ['foobar'],
unusedTaskTypes: ['barfoo'],
taskMaxAttempts: {
bar: customMaxAttempts,
foo: maxAttempts,
},
},
});
});

test('it claims tasks by setting their ownerId, status and retryAt', async () => {
const taskManagerId = uuid.v1();
const claimOwnershipUntil = new Date(Date.now());
Expand Down Expand Up @@ -1263,6 +1387,7 @@ if (doc['task.runAt'].size()!=0) {
logger: taskManagerLogger,
definitions,
excludedTaskTypes: [],
unusedTypes: [],
taskStore,
maxAttempts: 2,
getCapacity,
Expand Down
Loading