From 1578276708a6a036e736ef7c42f273fe94220e1d Mon Sep 17 00:00:00 2001 From: Heikki Hellgren Date: Wed, 22 Feb 2023 12:46:22 +0200 Subject: [PATCH 1/2] feat: add functionality to get scheduled tasks add new functionality to PluginTaskScheduler to return local and global tasks. this is to be able to trigger those tasks manually using the triggerTask functionality when there's no clear way to figure out the correct id to use as they in many cases come from a plugin. this is also usable for the upcoming debug plugin, see #9737 Signed-off-by: Heikki Hellgren --- .changeset/fresh-hairs-switch.md | 5 +++ packages/backend-tasks/api-report.md | 5 +++ .../src/tasks/PluginTaskSchedulerImpl.test.ts | 41 +++++++++++++++++++ .../src/tasks/PluginTaskSchedulerImpl.ts | 8 ++++ packages/backend-tasks/src/tasks/index.ts | 1 + packages/backend-tasks/src/tasks/types.ts | 21 ++++++++++ 6 files changed, 81 insertions(+) create mode 100644 .changeset/fresh-hairs-switch.md diff --git a/.changeset/fresh-hairs-switch.md b/.changeset/fresh-hairs-switch.md new file mode 100644 index 0000000000..16d3b119e5 --- /dev/null +++ b/.changeset/fresh-hairs-switch.md @@ -0,0 +1,5 @@ +--- +'@backstage/backend-tasks': minor +--- + +add functionality to get descriptions from the scheduler for triggering diff --git a/packages/backend-tasks/api-report.md b/packages/backend-tasks/api-report.md index 44ec4addbe..66f901adad 100644 --- a/packages/backend-tasks/api-report.md +++ b/packages/backend-tasks/api-report.md @@ -16,6 +16,7 @@ export type HumanDuration = HumanDuration_2; // @public export interface PluginTaskScheduler { createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner; + getScheduledTasks(): TaskDescriptor[]; scheduleTask( task: TaskScheduleDefinition & TaskInvocationDefinition, ): Promise; @@ -27,6 +28,10 @@ export function readTaskScheduleDefinitionFromConfig( config: Config, ): TaskScheduleDefinition; +// @public +export type TaskDescriptor = TaskScheduleDefinition & + Exclude; + // @public export type TaskFunction = | ((abortSignal: AbortSignal) => void | Promise) diff --git a/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.test.ts b/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.test.ts index 1847370d2b..d64f353c27 100644 --- a/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.test.ts +++ b/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.test.ts @@ -304,6 +304,47 @@ describe('PluginTaskManagerImpl', () => { ); }); + describe('can fetch task ids', () => { + it.each(databases.eachSupportedId())( + 'can fetch both global and local task ids, %p', + async databaseId => { + const { manager } = await init(databaseId); + + const fn = jest.fn(); + const promise = new Promise(resolve => fn.mockImplementation(resolve)); + await manager.scheduleTask({ + id: 'task1', + timeout: Duration.fromMillis(5000), + frequency: Duration.fromMillis(5000), + fn, + scope: 'global', + }); + + await manager.scheduleTask({ + id: 'task2', + timeout: Duration.fromMillis(5000), + frequency: Duration.fromMillis(5000), + fn, + scope: 'local', + }); + + await promise; + + const tasks = manager.getScheduledTasks(); + expect(tasks.length).toEqual(2); + expect(tasks[0].id).toEqual('task1'); + expect(tasks[1].id).toEqual('task2'); + expect(tasks[0].scope).toEqual('global'); + expect(tasks[1].scope).toEqual('local'); + expect(tasks[0].fn).toBeUndefined(); + expect(tasks[1].fn).toBeUndefined(); + expect(tasks[0].signal).toBeUndefined(); + expect(tasks[1].signal).toBeUndefined(); + }, + 60_000, + ); + }); + describe('parseDuration', () => { it('should parse durations', () => { expect(parseDuration({ milliseconds: 5000 })).toEqual('PT5S'); diff --git a/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts b/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts index e57ec93c1c..bbc2b45718 100644 --- a/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts +++ b/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts @@ -21,6 +21,7 @@ import { LocalTaskWorker } from './LocalTaskWorker'; import { TaskWorker } from './TaskWorker'; import { PluginTaskScheduler, + TaskDescriptor, TaskInvocationDefinition, TaskRunner, TaskScheduleDefinition, @@ -32,6 +33,7 @@ import { validateId } from './util'; */ export class PluginTaskSchedulerImpl implements PluginTaskScheduler { private readonly localTasksById = new Map(); + private readonly allScheduledTasks: TaskDescriptor[] = []; constructor( private readonly databaseFactory: () => Promise, @@ -94,6 +96,8 @@ export class PluginTaskSchedulerImpl implements PluginTaskScheduler { this.localTasksById.set(task.id, worker); } + const { fn: _, signal: __, ...descriptor } = task; + this.allScheduledTasks.push(descriptor as TaskDescriptor); } createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner { @@ -103,6 +107,10 @@ export class PluginTaskSchedulerImpl implements PluginTaskScheduler { }, }; } + + getScheduledTasks(): TaskDescriptor[] { + return this.allScheduledTasks; + } } export function parseDuration( diff --git a/packages/backend-tasks/src/tasks/index.ts b/packages/backend-tasks/src/tasks/index.ts index f2037ac9dd..6f9ac6d859 100644 --- a/packages/backend-tasks/src/tasks/index.ts +++ b/packages/backend-tasks/src/tasks/index.ts @@ -19,6 +19,7 @@ export { TaskScheduler } from './TaskScheduler'; export type { PluginTaskScheduler, TaskFunction, + TaskDescriptor, TaskInvocationDefinition, TaskRunner, TaskScheduleDefinition, diff --git a/packages/backend-tasks/src/tasks/types.ts b/packages/backend-tasks/src/tasks/types.ts index 6ec5165aec..d8aa58bcb3 100644 --- a/packages/backend-tasks/src/tasks/types.ts +++ b/packages/backend-tasks/src/tasks/types.ts @@ -31,6 +31,14 @@ export type TaskFunction = | ((abortSignal: AbortSignal) => void | Promise) | (() => void | Promise); +/** + * A type to describe a scheduled task. + * + * @public + */ +export type TaskDescriptor = TaskScheduleDefinition & + Exclude; + /** * Options that control the scheduling of a task. * @@ -307,6 +315,19 @@ export interface PluginTaskScheduler { * @param schedule - The task schedule */ createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner; + + /** + * Returns all scheduled tasks registered to this scheduler. + * + * @remarks + * + * This method is useful for triggering tasks manually using the triggerTask + * functionality. Note that the returned tasks contain only tasks that have + * been initialized in this instance of the scheduler. + * + * @returns Scheduled tasks + */ + getScheduledTasks(): TaskDescriptor[]; } function isValidOptionalDurationString(d: string | undefined): boolean { From ddd67399de0c3ef2e847af90f8e1636f122d972a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Tue, 7 Mar 2023 12:46:54 +0100 Subject: [PATCH 2/2] add migrations test, return tasks async MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- packages/backend-tasks/api-report.md | 12 ++- packages/backend-tasks/src/migrations.test.ts | 82 +++++++++++++++++++ .../src/tasks/LocalTaskWorker.ts | 4 +- .../src/tasks/PluginTaskSchedulerImpl.test.ts | 27 +++--- .../src/tasks/PluginTaskSchedulerImpl.ts | 52 +++++------- packages/backend-tasks/src/tasks/types.ts | 24 ++++-- 6 files changed, 148 insertions(+), 53 deletions(-) create mode 100644 packages/backend-tasks/src/migrations.test.ts diff --git a/packages/backend-tasks/api-report.md b/packages/backend-tasks/api-report.md index 66f901adad..1d978068b3 100644 --- a/packages/backend-tasks/api-report.md +++ b/packages/backend-tasks/api-report.md @@ -7,6 +7,7 @@ import { Config } from '@backstage/config'; import { DatabaseManager } from '@backstage/backend-common'; import { Duration } from 'luxon'; import { HumanDuration as HumanDuration_2 } from '@backstage/types'; +import { JsonObject } from '@backstage/types'; import { Logger } from 'winston'; import { PluginDatabaseManager } from '@backstage/backend-common'; @@ -16,7 +17,7 @@ export type HumanDuration = HumanDuration_2; // @public export interface PluginTaskScheduler { createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner; - getScheduledTasks(): TaskDescriptor[]; + getScheduledTasks(): Promise; scheduleTask( task: TaskScheduleDefinition & TaskInvocationDefinition, ): Promise; @@ -29,8 +30,13 @@ export function readTaskScheduleDefinitionFromConfig( ): TaskScheduleDefinition; // @public -export type TaskDescriptor = TaskScheduleDefinition & - Exclude; +export type TaskDescriptor = { + id: string; + scope: 'global' | 'local'; + settings: { + version: number; + } & JsonObject; +}; // @public export type TaskFunction = diff --git a/packages/backend-tasks/src/migrations.test.ts b/packages/backend-tasks/src/migrations.test.ts new file mode 100644 index 0000000000..4d37b9d39e --- /dev/null +++ b/packages/backend-tasks/src/migrations.test.ts @@ -0,0 +1,82 @@ +/* + * Copyright 2022 The Backstage Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { Knex } from 'knex'; +import { TestDatabases } from '@backstage/backend-test-utils'; +import fs from 'fs'; + +const migrationsDir = `${__dirname}/../migrations`; +const migrationsFiles = fs.readdirSync(migrationsDir).sort(); + +async function migrateUpOnce(knex: Knex): Promise { + await knex.migrate.up({ directory: migrationsDir }); +} + +async function migrateDownOnce(knex: Knex): Promise { + await knex.migrate.down({ directory: migrationsDir }); +} + +async function migrateUntilBefore(knex: Knex, target: string): Promise { + const index = migrationsFiles.indexOf(target); + if (index === -1) { + throw new Error(`Migration ${target} not found`); + } + for (let i = 0; i < index; i++) { + await migrateUpOnce(knex); + } +} + +describe('migrations', () => { + const databases = TestDatabases.create({ + ids: ['POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'], + }); + + it.each(databases.eachSupportedId())( + '20210928160613_init.js, %p', + async databaseId => { + const knex = await databases.init(databaseId); + + await migrateUntilBefore(knex, '20210928160613_init.js'); + await migrateUpOnce(knex); + + await knex('backstage_backend_tasks__tasks').insert({ + id: 'test', + settings_json: '{}', + next_run_start_at: knex.fn.now(), + }); + + await expect(knex('backstage_backend_tasks__tasks')).resolves.toEqual([ + { + id: 'test', + settings_json: '{}', + next_run_start_at: expect.anything(), + current_run_ticket: null, + current_run_started_at: null, + current_run_expires_at: null, + }, + ]); + + await migrateDownOnce(knex); + + await expect(knex('backstage_backend_tasks__tasks')).rejects.toThrow( + /backstage_backend_tasks__tasks/, + ); + + await knex.destroy(); + }, + 60_000, + ); +}); diff --git a/packages/backend-tasks/src/tasks/LocalTaskWorker.ts b/packages/backend-tasks/src/tasks/LocalTaskWorker.ts index 0edec5e84e..347d22e48e 100644 --- a/packages/backend-tasks/src/tasks/LocalTaskWorker.ts +++ b/packages/backend-tasks/src/tasks/LocalTaskWorker.ts @@ -39,8 +39,9 @@ export class LocalTaskWorker { this.logger.info( `Task worker starting: ${this.taskId}, ${JSON.stringify(settings)}`, ); - let attemptNum = 1; + (async () => { + let attemptNum = 1; for (;;) { try { if (settings.initialDelayDuration) { @@ -60,6 +61,7 @@ export class LocalTaskWorker { options?.signal, ); } + this.logger.info(`Task worker finished: ${this.taskId}`); attemptNum = 0; break; diff --git a/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.test.ts b/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.test.ts index d64f353c27..9a29810683 100644 --- a/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.test.ts +++ b/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.test.ts @@ -309,9 +309,8 @@ describe('PluginTaskManagerImpl', () => { 'can fetch both global and local task ids, %p', async databaseId => { const { manager } = await init(databaseId); - const fn = jest.fn(); - const promise = new Promise(resolve => fn.mockImplementation(resolve)); + await manager.scheduleTask({ id: 'task1', timeout: Duration.fromMillis(5000), @@ -328,18 +327,18 @@ describe('PluginTaskManagerImpl', () => { scope: 'local', }); - await promise; - - const tasks = manager.getScheduledTasks(); - expect(tasks.length).toEqual(2); - expect(tasks[0].id).toEqual('task1'); - expect(tasks[1].id).toEqual('task2'); - expect(tasks[0].scope).toEqual('global'); - expect(tasks[1].scope).toEqual('local'); - expect(tasks[0].fn).toBeUndefined(); - expect(tasks[1].fn).toBeUndefined(); - expect(tasks[0].signal).toBeUndefined(); - expect(tasks[1].signal).toBeUndefined(); + await expect(manager.getScheduledTasks()).resolves.toEqual([ + { + id: 'task1', + scope: 'global', + settings: expect.objectContaining({ cadence: 'PT5S' }), + }, + { + id: 'task2', + scope: 'local', + settings: expect.objectContaining({ cadence: 'PT5S' }), + }, + ]); }, 60_000, ); diff --git a/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts b/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts index bbc2b45718..253d7869df 100644 --- a/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts +++ b/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts @@ -25,6 +25,7 @@ import { TaskInvocationDefinition, TaskRunner, TaskScheduleDefinition, + TaskSettingsV2, } from './types'; import { validateId } from './util'; @@ -57,6 +58,14 @@ export class PluginTaskSchedulerImpl implements PluginTaskScheduler { validateId(task.id); const scope = task.scope ?? 'global'; + const settings: TaskSettingsV2 = { + version: 2, + cadence: parseDuration(task.frequency), + initialDelayDuration: + task.initialDelay && parseDuration(task.initialDelay), + timeoutAfterDuration: parseDuration(task.timeout), + }; + if (scope === 'global') { const knex = await this.databaseFactory(); const worker = new TaskWorker( @@ -65,39 +74,22 @@ export class PluginTaskSchedulerImpl implements PluginTaskScheduler { knex, this.logger.child({ task: task.id }), ); - - await worker.start( - { - version: 2, - cadence: parseDuration(task.frequency), - initialDelayDuration: - task.initialDelay && parseDuration(task.initialDelay), - timeoutAfterDuration: parseDuration(task.timeout), - }, - { - signal: task.signal, - }, - ); + await worker.start(settings, { signal: task.signal }); } else { - const worker = new LocalTaskWorker(task.id, task.fn, this.logger); - - worker.start( - { - version: 2, - cadence: parseDuration(task.frequency), - initialDelayDuration: - task.initialDelay && parseDuration(task.initialDelay), - timeoutAfterDuration: parseDuration(task.timeout), - }, - { - signal: task.signal, - }, + const worker = new LocalTaskWorker( + task.id, + task.fn, + this.logger.child({ task: task.id }), ); - + worker.start(settings, { signal: task.signal }); this.localTasksById.set(task.id, worker); } - const { fn: _, signal: __, ...descriptor } = task; - this.allScheduledTasks.push(descriptor as TaskDescriptor); + + this.allScheduledTasks.push({ + id: task.id, + scope: scope, + settings: settings, + }); } createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner { @@ -108,7 +100,7 @@ export class PluginTaskSchedulerImpl implements PluginTaskScheduler { }; } - getScheduledTasks(): TaskDescriptor[] { + async getScheduledTasks(): Promise { return this.allScheduledTasks; } } diff --git a/packages/backend-tasks/src/tasks/types.ts b/packages/backend-tasks/src/tasks/types.ts index d8aa58bcb3..14339dda0e 100644 --- a/packages/backend-tasks/src/tasks/types.ts +++ b/packages/backend-tasks/src/tasks/types.ts @@ -14,7 +14,7 @@ * limitations under the License. */ -import { HumanDuration } from '@backstage/types'; +import { HumanDuration, JsonObject } from '@backstage/types'; import { CronTime } from 'cron'; import { Duration } from 'luxon'; import { z } from 'zod'; @@ -32,12 +32,26 @@ export type TaskFunction = | (() => void | Promise); /** - * A type to describe a scheduled task. + * A semi-opaque type to describe an actively scheduled task. * * @public */ -export type TaskDescriptor = TaskScheduleDefinition & - Exclude; +export type TaskDescriptor = { + /** + * The unique identifier of the task. + */ + id: string; + /** + * The scope of the task. + */ + scope: 'global' | 'local'; + /** + * The settings that control the task flow. This is a semi-opaque structure + * that is mainly there for debugging purposes. Do not make any assumptions + * about the contents of this field. + */ + settings: { version: number } & JsonObject; +}; /** * Options that control the scheduling of a task. @@ -327,7 +341,7 @@ export interface PluginTaskScheduler { * * @returns Scheduled tasks */ - getScheduledTasks(): TaskDescriptor[]; + getScheduledTasks(): Promise; } function isValidOptionalDurationString(d: string | undefined): boolean {