From cfd779a9bc165d109cdf50adc22a1de404de8727 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Thu, 21 Apr 2022 14:18:15 +0200 Subject: [PATCH] Add local task running support to TaskScheduler MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- .changeset/swift-parrots-hammer.md | 5 + packages/backend-tasks/api-report.md | 4 + .../src/tasks/LocalTaskWorker.test.ts | 105 +++++++++++++ .../src/tasks/LocalTaskWorker.ts | 138 ++++++++++++++++++ .../src/tasks/PluginTaskSchedulerImpl.ts | 63 +++++--- .../backend-tasks/src/tasks/TaskWorker.ts | 25 +++- packages/backend-tasks/src/tasks/types.ts | 30 ++++ 7 files changed, 348 insertions(+), 22 deletions(-) create mode 100644 .changeset/swift-parrots-hammer.md create mode 100644 packages/backend-tasks/src/tasks/LocalTaskWorker.test.ts create mode 100644 packages/backend-tasks/src/tasks/LocalTaskWorker.ts diff --git a/.changeset/swift-parrots-hammer.md b/.changeset/swift-parrots-hammer.md new file mode 100644 index 0000000000..1c7c178bdb --- /dev/null +++ b/.changeset/swift-parrots-hammer.md @@ -0,0 +1,5 @@ +--- +'@backstage/backend-tasks': patch +--- + +The task scheduler now has two new methods: `scheduleLocalTask` and `createScheduledLocalTaskRunner`, which can be used to perform local-only recurring tasks. This can be used to replace usages of `runPeriodically` helpers. diff --git a/packages/backend-tasks/api-report.md b/packages/backend-tasks/api-report.md index 9f02088bdf..e884b95a0d 100644 --- a/packages/backend-tasks/api-report.md +++ b/packages/backend-tasks/api-report.md @@ -11,7 +11,11 @@ import { Logger } from 'winston'; // @public export interface PluginTaskScheduler { + createScheduledLocalTaskRunner(schedule: TaskScheduleDefinition): TaskRunner; createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner; + scheduleLocalTask( + task: TaskScheduleDefinition & TaskInvocationDefinition, + ): Promise; scheduleTask( task: TaskScheduleDefinition & TaskInvocationDefinition, ): Promise; diff --git a/packages/backend-tasks/src/tasks/LocalTaskWorker.test.ts b/packages/backend-tasks/src/tasks/LocalTaskWorker.test.ts new file mode 100644 index 0000000000..bdbd976def --- /dev/null +++ b/packages/backend-tasks/src/tasks/LocalTaskWorker.test.ts @@ -0,0 +1,105 @@ +/* + * 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 { getVoidLogger } from '@backstage/backend-common'; +import { AbortController } from 'node-abort-controller'; +import { LocalTaskWorker } from './LocalTaskWorker'; + +describe('LocalTaskWorker', () => { + const logger = getVoidLogger(); + + it('runs the happy path (with iso duration) and handles cancellation', async () => { + const fn = jest.fn(); + const controller = new AbortController(); + + const worker = new LocalTaskWorker('a', fn, logger); + worker.start( + { + version: 2, + initialDelayDuration: 'PT0.2S', + cadence: 'PT0.2S', + timeoutAfterDuration: 'PT1S', + }, + { signal: controller.signal }, + ); + + // TODO(freben): Rewrite to fake timers - tried, but it wouldn't work + expect(fn).toBeCalledTimes(0); + await new Promise(r => setTimeout(r, 100)); + expect(fn).toBeCalledTimes(0); + await new Promise(r => setTimeout(r, 200)); + expect(fn).toBeCalledTimes(1); + await new Promise(r => setTimeout(r, 200)); + expect(fn).toBeCalledTimes(2); + controller.abort(); + await new Promise(r => setTimeout(r, 200)); + expect(fn).toBeCalledTimes(2); + }); + + it('runs the happy path (with a cron expression) and handles cancellation', async () => { + const fn = jest.fn(); + const controller = new AbortController(); + + // Await until system time is just past a second boundary (since cron is + // wall clock based) + await new Promise(r => setTimeout(r, 1000 - (Date.now() % 1000) + 10)); + + const worker = new LocalTaskWorker('a', fn, logger); + worker.start( + { + version: 2, + initialDelayDuration: 'PT0.2S', + cadence: '* * * * * *', + timeoutAfterDuration: 'PT1S', + }, + { signal: controller.signal }, + ); + + // TODO(freben): Rewrite to fake timers - tried, but it wouldn't work + expect(fn).toBeCalledTimes(0); + await new Promise(r => setTimeout(r, 100)); + expect(fn).toBeCalledTimes(0); + await new Promise(r => setTimeout(r, 200)); + expect(fn).toBeCalledTimes(1); + await new Promise(r => setTimeout(r, 1000)); + expect(fn).toBeCalledTimes(2); + controller.abort(); + await new Promise(r => setTimeout(r, 1000)); + expect(fn).toBeCalledTimes(2); + }); + + it('can trigger to abort wait', async () => { + const fn = jest.fn(); + + const worker = new LocalTaskWorker('a', fn, logger); + worker.start({ + version: 2, + initialDelayDuration: 'PT0.2S', + cadence: 'PT0.2S', + timeoutAfterDuration: 'PT1S', + }); + + // TODO(freben): Rewrite to fake timers - tried, but it wouldn't work + expect(fn).toBeCalledTimes(0); + await new Promise(r => setTimeout(r, 100)); + expect(fn).toBeCalledTimes(0); + await new Promise(r => setTimeout(r, 200)); + expect(fn).toBeCalledTimes(1); + worker.trigger(); + await new Promise(r => setTimeout(r, 10)); + expect(fn).toBeCalledTimes(2); + }); +}); diff --git a/packages/backend-tasks/src/tasks/LocalTaskWorker.ts b/packages/backend-tasks/src/tasks/LocalTaskWorker.ts new file mode 100644 index 0000000000..dd6208df1a --- /dev/null +++ b/packages/backend-tasks/src/tasks/LocalTaskWorker.ts @@ -0,0 +1,138 @@ +/* + * 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 { ConflictError } from '@backstage/errors'; +import { CronTime } from 'cron'; +import { DateTime, Duration } from 'luxon'; +import { AbortController, AbortSignal } from 'node-abort-controller'; +import { Logger } from 'winston'; +import { TaskFunction, TaskSettingsV2 } from './types'; +import { delegateAbortController, sleep } from './util'; + +export class LocalTaskWorker { + private abortWait: AbortController | undefined; + + constructor( + private readonly taskId: string, + private readonly fn: TaskFunction, + private readonly logger: Logger, + ) {} + + get id(): string { + return this.taskId; + } + + start(settings: TaskSettingsV2, options?: { signal?: AbortSignal }) { + this.logger.info( + `Task worker starting: ${this.taskId}, ${JSON.stringify(settings)}`, + ); + + (async () => { + try { + if (settings.initialDelayDuration) { + await sleep( + Duration.fromISO(settings.initialDelayDuration), + options?.signal, + ); + } + + while (!options?.signal?.aborted) { + const startTime = Date.now(); + await this.runOnce(settings, options?.signal); + const endTime = Date.now(); + await this.waitUntilNext( + settings, + endTime - startTime, + options?.signal, + ); + } + + this.logger.info(`Task worker finished: ${this.taskId}`); + } catch (e) { + this.logger.warn(`Task worker failed unexpectedly, ${e}`); + } + })(); + } + + trigger(): void { + if (!this.abortWait) { + throw new ConflictError(`Task ${this.taskId} is currently running`); + } + this.abortWait.abort(); + } + + /** + * Makes a single attempt at running the task to completion. + */ + private async runOnce( + settings: TaskSettingsV2, + signal?: AbortSignal, + ): Promise { + // Abort the task execution either if the worker is stopped, or if the + // task timeout is hit + const taskAbortController = delegateAbortController(signal); + const timeoutHandle = setTimeout(() => { + taskAbortController.abort(); + }, Duration.fromISO(settings.timeoutAfterDuration).as('milliseconds')); + + try { + await this.fn(taskAbortController.signal); + } catch (e) { + // ignore intentionally + } + + // release resources + clearTimeout(timeoutHandle); + taskAbortController.abort(); + } + + /** + * Sleeps until it's time to run the task again. + */ + private async waitUntilNext( + settings: TaskSettingsV2, + lastRunMillis: number, + signal?: AbortSignal, + ) { + if (signal?.aborted) { + return; + } + + const isCron = !settings.cadence.startsWith('P'); + let dt: number; + + if (isCron) { + const nextRun = +new CronTime(settings.cadence).sendAt().toDate(); + dt = nextRun - Date.now(); + } else { + dt = + Duration.fromISO(settings.cadence).as('milliseconds') - lastRunMillis; + } + + dt = Math.max(dt, 0); + + this.logger.debug( + `task: ${this.taskId} will next occur around ${DateTime.now().plus( + Duration.fromMillis(dt), + )}`, + ); + + this.abortWait = delegateAbortController(signal); + await sleep(Duration.fromMillis(dt), this.abortWait.signal); + this.abortWait.abort(); // cleans up resources + this.abortWait = undefined; + } +} diff --git a/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts b/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts index 44f46e7237..f9a4ca77bf 100644 --- a/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts +++ b/packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts @@ -16,6 +16,7 @@ import { Knex } from 'knex'; import { Logger } from 'winston'; +import { LocalTaskWorker } from './LocalTaskWorker'; import { TaskWorker } from './TaskWorker'; import { PluginTaskScheduler, @@ -24,38 +25,27 @@ import { TaskScheduleDefinition, } from './types'; import { validateId } from './util'; -import { DB_TASKS_TABLE, DbTasksRow } from '../database/tables'; -import { ConflictError, NotFoundError } from '@backstage/errors'; /** * Implements the actual task management. */ export class PluginTaskSchedulerImpl implements PluginTaskScheduler { + private readonly localTasksById = new Map(); + constructor( private readonly databaseFactory: () => Promise, private readonly logger: Logger, ) {} async triggerTask(id: string): Promise { + const localTask = this.localTasksById.get(id); + if (localTask) { + localTask.trigger(); + return; + } + const knex = await this.databaseFactory(); - - // check if task exists - const rows = await knex(DB_TASKS_TABLE) - .select(knex.raw(1)) - .where('id', '=', id); - if (rows.length !== 1) { - throw new NotFoundError(`Task ${id} does not exist`); - } - - const updatedRows = await knex(DB_TASKS_TABLE) - .where('id', '=', id) - .whereNull('current_run_ticket') - .update({ - next_run_start_at: knex.fn.now(), - }); - if (updatedRows < 1) { - throw new ConflictError(`Task ${id} is currently running`); - } + await TaskWorker.trigger(knex, id); } async scheduleTask( @@ -82,6 +72,31 @@ export class PluginTaskSchedulerImpl implements PluginTaskScheduler { ); } + async scheduleLocalTask( + task: TaskScheduleDefinition & TaskInvocationDefinition, + ): Promise { + validateId(task.id); + + const worker = new LocalTaskWorker(task.id, task.fn, this.logger); + + worker.start( + { + version: 2, + cadence: + 'cron' in task.frequency + ? task.frequency.cron + : task.frequency.toISO(), + initialDelayDuration: task.initialDelay?.toISO(), + timeoutAfterDuration: task.timeout.toISO(), + }, + { + signal: task.signal, + }, + ); + + this.localTasksById.set(task.id, worker); + } + createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner { return { run: async task => { @@ -89,4 +104,12 @@ export class PluginTaskSchedulerImpl implements PluginTaskScheduler { }, }; } + + createScheduledLocalTaskRunner(schedule: TaskScheduleDefinition): TaskRunner { + return { + run: async task => { + await this.scheduleLocalTask({ ...task, ...schedule }); + }, + }; + } } diff --git a/packages/backend-tasks/src/tasks/TaskWorker.ts b/packages/backend-tasks/src/tasks/TaskWorker.ts index 02bc115ddd..0656746d37 100644 --- a/packages/backend-tasks/src/tasks/TaskWorker.ts +++ b/packages/backend-tasks/src/tasks/TaskWorker.ts @@ -14,6 +14,8 @@ * limitations under the License. */ +import { ConflictError, NotFoundError } from '@backstage/errors'; +import { CronTime } from 'cron'; import { Knex } from 'knex'; import { DateTime, Duration } from 'luxon'; import { AbortSignal } from 'node-abort-controller'; @@ -22,7 +24,6 @@ import { Logger } from 'winston'; import { DbTasksRow, DB_TASKS_TABLE } from '../database/tables'; import { TaskFunction, TaskSettingsV2, taskSettingsV2Schema } from './types'; import { delegateAbortController, nowPlus, sleep } from './util'; -import { CronTime } from 'cron'; const DEFAULT_WORK_CHECK_FREQUENCY = Duration.fromObject({ seconds: 5 }); @@ -76,12 +77,32 @@ export class TaskWorker { })(); } + static async trigger(knex: Knex, taskId: string): Promise { + // check if task exists + const rows = await knex(DB_TASKS_TABLE) + .select(knex.raw(1)) + .where('id', '=', taskId); + if (rows.length !== 1) { + throw new NotFoundError(`Task ${taskId} does not exist`); + } + + const updatedRows = await knex(DB_TASKS_TABLE) + .where('id', '=', taskId) + .whereNull('current_run_ticket') + .update({ + next_run_start_at: knex.fn.now(), + }); + if (updatedRows < 1) { + throw new ConflictError(`Task ${taskId} is currently running`); + } + } + /** * Makes a single attempt at running the task to completion, if ready. * * @returns The outcome of the attempt */ - async runOnce( + private async runOnce( signal?: AbortSignal, ): Promise< | { result: 'not-ready-yet' } diff --git a/packages/backend-tasks/src/tasks/types.ts b/packages/backend-tasks/src/tasks/types.ts index 1e09e52cd0..39ac2eca6e 100644 --- a/packages/backend-tasks/src/tasks/types.ts +++ b/packages/backend-tasks/src/tasks/types.ts @@ -174,6 +174,21 @@ export interface PluginTaskScheduler { task: TaskScheduleDefinition & TaskInvocationDefinition, ): Promise; + /** + * Schedules a task function for running only locally on this machine, without + * any form of coordination, locking or concurrency control across hosts. This + * convenience method performs both the scheduling and invocation in one go. + * + * @remarks + * + * If a task with this ID was already registered, a ConflictError is thrown. + * + * @param task - The task definition + */ + scheduleLocalTask( + task: TaskScheduleDefinition & TaskInvocationDefinition, + ): Promise; + /** * Creates a scheduled but dormant recurring task, ready to be launched at a * later time. @@ -187,6 +202,21 @@ export interface PluginTaskScheduler { * @param schedule - The task schedule */ createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner; + + /** + * Creates a local (without any form of coordination, locking or concurrency + * control across hosts) scheduled but dormant recurring task, ready to be + * launched at a later time. + * + * @remarks + * + * This method is useful for pre-creating a schedule in outer code to be + * passed into an inner implementation, such that the outer code controls + * scheduling while inner code controls implementation. + * + * @param schedule - The task schedule + */ + createScheduledLocalTaskRunner(schedule: TaskScheduleDefinition): TaskRunner; } function isValidOptionalDurationString(d: string | undefined): boolean {