From 906705b77be1be59177fac41e5b79b9b6b18d5f9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Thu, 25 Apr 2024 16:06:57 +0200 Subject: [PATCH] arrange in backend-plugin-api and backend-defaults MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- .changeset/calm-cars-serve.md | 2 +- packages/backend-app-api/api-report.md | 4 +- .../services/implementations}/deprecated.ts | 8 +- .../src/services/implementations/index.ts | 3 +- .../scheduler/schedulerServiceFactory.ts | 5 +- packages/backend-common/api-report.md | 2 +- .../backend-defaults/api-report-scheduler.md | 16 + .../scheduler/20210928160613_init.js | 64 +++ packages/backend-defaults/package.json | 33 +- .../backend-defaults/src/CreateBackend.ts | 2 +- .../scheduler/database/migrateBackendTasks.ts | 31 ++ .../entrypoints/scheduler/database/tables.ts | 27 + .../src/entrypoints/scheduler/index.ts | 11 +- .../scheduler/lib/LocalTaskWorker.test.ts | 109 ++++ .../scheduler/lib/LocalTaskWorker.ts | 155 ++++++ .../lib/PluginTaskSchedulerImpl.test.ts | 349 ++++++++++++ .../scheduler/lib/PluginTaskSchedulerImpl.ts | 169 ++++++ .../lib/PluginTaskSchedulerJanitor.test.ts | 96 ++++ .../lib/PluginTaskSchedulerJanitor.ts | 96 ++++ .../scheduler/lib/TaskScheduler.test.ts | 84 +++ .../scheduler/lib/TaskScheduler.ts | 98 ++++ .../scheduler/lib/TaskWorker.test.ts | 507 ++++++++++++++++++ .../entrypoints/scheduler/lib/TaskWorker.ts | 373 +++++++++++++ .../__testUtils__/createTestScopedSignal.ts | 28 + .../src/entrypoints/scheduler/lib/types.ts | 158 ++++++ .../entrypoints/scheduler/lib/util.test.ts | 113 ++++ .../src/entrypoints/scheduler/lib/util.ts | 111 ++++ .../scheduler/schedulerServiceFactory.test.ts | 46 ++ .../scheduler/schedulerServiceFactory.ts | 42 ++ packages/backend-defaults/src/setupTests.ts | 25 + .../api-report-scheduler.md | 79 --- packages/backend-plugin-api/api-report.md | 76 ++- packages/backend-plugin-api/package.json | 4 - .../readTaskScheduleDefinitionFromConfig.ts | 78 --- packages/backend-plugin-api/src/index.ts | 1 - .../definitions/SchedulerService.test.ts} | 36 +- .../definitions/SchedulerService.ts} | 85 ++- .../src/services/definitions/coreServices.ts | 2 +- .../src/services/definitions/index.ts | 10 + .../backend-tasks/src/tasks/TaskScheduler.ts | 2 +- .../readTaskScheduleDefinitionFromConfig.ts | 2 +- packages/backend-tasks/src/tasks/types.ts | 14 +- packages/backend-test-utils/api-report.md | 2 +- yarn.lock | 10 + 44 files changed, 2938 insertions(+), 230 deletions(-) rename packages/{backend-plugin-api/src => backend-app-api/src/services/implementations}/deprecated.ts (70%) create mode 100644 packages/backend-defaults/api-report-scheduler.md create mode 100644 packages/backend-defaults/migrations/scheduler/20210928160613_init.js create mode 100644 packages/backend-defaults/src/entrypoints/scheduler/database/migrateBackendTasks.ts create mode 100644 packages/backend-defaults/src/entrypoints/scheduler/database/tables.ts rename packages/{backend-plugin-api => backend-defaults}/src/entrypoints/scheduler/index.ts (68%) create mode 100644 packages/backend-defaults/src/entrypoints/scheduler/lib/LocalTaskWorker.test.ts create mode 100644 packages/backend-defaults/src/entrypoints/scheduler/lib/LocalTaskWorker.ts create mode 100644 packages/backend-defaults/src/entrypoints/scheduler/lib/PluginTaskSchedulerImpl.test.ts create mode 100644 packages/backend-defaults/src/entrypoints/scheduler/lib/PluginTaskSchedulerImpl.ts create mode 100644 packages/backend-defaults/src/entrypoints/scheduler/lib/PluginTaskSchedulerJanitor.test.ts create mode 100644 packages/backend-defaults/src/entrypoints/scheduler/lib/PluginTaskSchedulerJanitor.ts create mode 100644 packages/backend-defaults/src/entrypoints/scheduler/lib/TaskScheduler.test.ts create mode 100644 packages/backend-defaults/src/entrypoints/scheduler/lib/TaskScheduler.ts create mode 100644 packages/backend-defaults/src/entrypoints/scheduler/lib/TaskWorker.test.ts create mode 100644 packages/backend-defaults/src/entrypoints/scheduler/lib/TaskWorker.ts create mode 100644 packages/backend-defaults/src/entrypoints/scheduler/lib/__testUtils__/createTestScopedSignal.ts create mode 100644 packages/backend-defaults/src/entrypoints/scheduler/lib/types.ts create mode 100644 packages/backend-defaults/src/entrypoints/scheduler/lib/util.test.ts create mode 100644 packages/backend-defaults/src/entrypoints/scheduler/lib/util.ts create mode 100644 packages/backend-defaults/src/entrypoints/scheduler/schedulerServiceFactory.test.ts create mode 100644 packages/backend-defaults/src/entrypoints/scheduler/schedulerServiceFactory.ts create mode 100644 packages/backend-defaults/src/setupTests.ts delete mode 100644 packages/backend-plugin-api/api-report-scheduler.md delete mode 100644 packages/backend-plugin-api/src/entrypoints/scheduler/readTaskScheduleDefinitionFromConfig.ts rename packages/backend-plugin-api/src/{entrypoints/scheduler/readTaskScheduleDefinitionFromConfig.test.ts => services/definitions/SchedulerService.test.ts} (76%) rename packages/backend-plugin-api/src/{entrypoints/scheduler/types.ts => services/definitions/SchedulerService.ts} (81%) diff --git a/.changeset/calm-cars-serve.md b/.changeset/calm-cars-serve.md index 5197edda5c..5df2a18cd4 100644 --- a/.changeset/calm-cars-serve.md +++ b/.changeset/calm-cars-serve.md @@ -2,4 +2,4 @@ '@backstage/backend-tasks': patch --- -Marked all exports as deprecated and pointed at `@backstage/backend-plugin-api/scheduler` and `@backstage/backend-defaults/scheduler` +Marked all exports as deprecated and pointed at `@backstage/backend-plugin-api` and `@backstage/backend-defaults` diff --git a/packages/backend-app-api/api-report.md b/packages/backend-app-api/api-report.md index cb0469d9d7..789873e53c 100644 --- a/packages/backend-app-api/api-report.md +++ b/packages/backend-app-api/api-report.md @@ -36,7 +36,7 @@ import { RootConfigService } from '@backstage/backend-plugin-api'; import { RootHttpRouterService } from '@backstage/backend-plugin-api'; import { RootLifecycleService } from '@backstage/backend-plugin-api'; import { RootLoggerService } from '@backstage/backend-plugin-api'; -import { SchedulerService } from '@backstage/backend-plugin-api/scheduler'; +import { SchedulerService } from '@backstage/backend-plugin-api'; import type { Server } from 'node:http'; import { ServiceFactory } from '@backstage/backend-plugin-api'; import { ServiceFactoryOrFunction } from '@backstage/backend-plugin-api'; @@ -326,7 +326,7 @@ export const rootLoggerServiceFactory: () => ServiceFactory< 'root' >; -// @public (undocumented) +// @public @deprecated (undocumented) export const schedulerServiceFactory: () => ServiceFactory< SchedulerService, 'plugin' diff --git a/packages/backend-plugin-api/src/deprecated.ts b/packages/backend-app-api/src/services/implementations/deprecated.ts similarity index 70% rename from packages/backend-plugin-api/src/deprecated.ts rename to packages/backend-app-api/src/services/implementations/deprecated.ts index 3cf6b0993e..6f4f76e679 100644 --- a/packages/backend-plugin-api/src/deprecated.ts +++ b/packages/backend-app-api/src/services/implementations/deprecated.ts @@ -14,10 +14,4 @@ * limitations under the License. */ -import { type SchedulerService as SchedulerService_ } from '@backstage/backend-plugin-api/scheduler'; - -/** - * @public - * @deprecated import from `@backstage/backend-plugin-api/scheduler` instead - */ -export type SchedulerService = SchedulerService_; +export * from './scheduler'; diff --git a/packages/backend-app-api/src/services/implementations/index.ts b/packages/backend-app-api/src/services/implementations/index.ts index a1114ab3e3..e6656801a9 100644 --- a/packages/backend-app-api/src/services/implementations/index.ts +++ b/packages/backend-app-api/src/services/implementations/index.ts @@ -28,7 +28,8 @@ export * from './permissions'; export * from './rootHttpRouter'; export * from './rootLifecycle'; export * from './rootLogger'; -export * from './scheduler'; export * from './tokenManager'; export * from './urlReader'; export * from './userInfo'; + +export * from './deprecated'; diff --git a/packages/backend-app-api/src/services/implementations/scheduler/schedulerServiceFactory.ts b/packages/backend-app-api/src/services/implementations/scheduler/schedulerServiceFactory.ts index dc9b3a9864..b4370761d9 100644 --- a/packages/backend-app-api/src/services/implementations/scheduler/schedulerServiceFactory.ts +++ b/packages/backend-app-api/src/services/implementations/scheduler/schedulerServiceFactory.ts @@ -20,7 +20,10 @@ import { } from '@backstage/backend-plugin-api'; import { TaskScheduler } from '@backstage/backend-tasks'; -/** @public */ +/** + * @public + * @deprecated Please import from `@backstage/backend-defaults/scheduler` instead. + */ export const schedulerServiceFactory = createServiceFactory({ service: coreServices.scheduler, deps: { diff --git a/packages/backend-common/api-report.md b/packages/backend-common/api-report.md index f9f25c3b81..42df32dc1d 100644 --- a/packages/backend-common/api-report.md +++ b/packages/backend-common/api-report.md @@ -59,7 +59,7 @@ import { resolvePackagePath as resolvePackagePath_2 } from '@backstage/backend-p import { resolveSafeChildPath as resolveSafeChildPath_2 } from '@backstage/backend-plugin-api'; import { RootConfigService } from '@backstage/backend-plugin-api'; import { Router } from 'express'; -import { SchedulerService } from '@backstage/backend-plugin-api/scheduler'; +import { SchedulerService } from '@backstage/backend-plugin-api'; import { SearchOptions } from '@backstage/backend-plugin-api'; import { SearchResponse } from '@backstage/backend-plugin-api'; import { SearchResponseFile } from '@backstage/backend-plugin-api'; diff --git a/packages/backend-defaults/api-report-scheduler.md b/packages/backend-defaults/api-report-scheduler.md new file mode 100644 index 0000000000..a2dad43d60 --- /dev/null +++ b/packages/backend-defaults/api-report-scheduler.md @@ -0,0 +1,16 @@ +## API Report File for "@backstage/backend-defaults" + +> Do not edit this file. It is a report generated by [API Extractor](https://api-extractor.com/). + +```ts +import { SchedulerService } from '@backstage/backend-plugin-api'; +import { ServiceFactory } from '@backstage/backend-plugin-api'; + +// @public +export const schedulerServiceFactory: () => ServiceFactory< + SchedulerService, + 'plugin' +>; + +// (No @packageDocumentation comment for this package) +``` diff --git a/packages/backend-defaults/migrations/scheduler/20210928160613_init.js b/packages/backend-defaults/migrations/scheduler/20210928160613_init.js new file mode 100644 index 0000000000..f9900cab11 --- /dev/null +++ b/packages/backend-defaults/migrations/scheduler/20210928160613_init.js @@ -0,0 +1,64 @@ +/* + * Copyright 2020 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. + */ + +// @ts-check + +/** + * @param {import('knex').Knex} knex + */ +exports.up = async function up(knex) { + // + // tasks + // + await knex.schema.createTable('backstage_backend_tasks__tasks', table => { + table.comment('Tasks used for scheduling work on multiple workers'); + table + .string('id') + .primary() + .notNullable() + .comment('The unique ID of this particular task'); + table + .text('settings_json') + .notNullable() + .comment('JSON serialized object with properties for this task'); + table + .dateTime('next_run_start_at') + .notNullable() + .comment('The next time that the task should be started'); + table + .text('current_run_ticket') + .nullable() + .comment('A unique ticket for the current task run'); + table + .dateTime('current_run_started_at') + .nullable() + .comment('The time that the current task run started'); + table + .dateTime('current_run_expires_at') + .nullable() + .comment('The time that the current task run will time out'); + }); +}; + +/** + * @param {import('knex').Knex} knex + */ +exports.down = async function down(knex) { + // + // tasks + // + await knex.schema.dropTable('backstage_backend_tasks__tasks'); +}; diff --git a/packages/backend-defaults/package.json b/packages/backend-defaults/package.json index b5da7abbcf..742a1f99c1 100644 --- a/packages/backend-defaults/package.json +++ b/packages/backend-defaults/package.json @@ -12,6 +12,21 @@ "backstage": { "role": "node-library" }, + "exports": { + ".": "./src/index.ts", + "./scheduler": "./src/entrypoints/scheduler/index.ts", + "./package.json": "./package.json" + }, + "typesVersions": { + "*": { + "scheduler": [ + "src/entrypoints/scheduler/index.ts" + ], + "package.json": [ + "package.json" + ] + } + }, "homepage": "https://backstage.io", "repository": { "type": "git", @@ -34,14 +49,26 @@ "dependencies": { "@backstage/backend-app-api": "workspace:^", "@backstage/backend-common": "workspace:^", - "@backstage/plugin-events-node": "workspace:^" + "@backstage/backend-plugin-api": "workspace:^", + "@backstage/errors": "workspace:^", + "@backstage/plugin-events-node": "workspace:^", + "@backstage/types": "workspace:^", + "@opentelemetry/api": "^1.3.0", + "cron": "^3.0.0", + "knex": "^3.0.0", + "lodash": "^4.17.21", + "luxon": "^3.0.0", + "uuid": "^9.0.0", + "zod": "^3.22.4" }, "devDependencies": { "@backstage/backend-plugin-api": "workspace:^", "@backstage/backend-test-utils": "workspace:^", - "@backstage/cli": "workspace:^" + "@backstage/cli": "workspace:^", + "wait-for-expect": "^3.0.2" }, "files": [ - "dist" + "dist", + "migrations" ] } diff --git a/packages/backend-defaults/src/CreateBackend.ts b/packages/backend-defaults/src/CreateBackend.ts index e9bdc03262..811f980d12 100644 --- a/packages/backend-defaults/src/CreateBackend.ts +++ b/packages/backend-defaults/src/CreateBackend.ts @@ -28,7 +28,6 @@ import { loggerServiceFactory, permissionsServiceFactory, rootLoggerServiceFactory, - schedulerServiceFactory, tokenManagerServiceFactory, urlReaderServiceFactory, identityServiceFactory, @@ -37,6 +36,7 @@ import { userInfoServiceFactory, } from '@backstage/backend-app-api'; import { eventsServiceFactory } from '@backstage/plugin-events-node'; +import { schedulerServiceFactory } from '@backstage/backend-defaults/scheduler'; export const defaultServiceFactories = [ authServiceFactory(), diff --git a/packages/backend-defaults/src/entrypoints/scheduler/database/migrateBackendTasks.ts b/packages/backend-defaults/src/entrypoints/scheduler/database/migrateBackendTasks.ts new file mode 100644 index 0000000000..959e6097f4 --- /dev/null +++ b/packages/backend-defaults/src/entrypoints/scheduler/database/migrateBackendTasks.ts @@ -0,0 +1,31 @@ +/* + * Copyright 2021 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 { resolvePackagePath } from '@backstage/backend-common'; +import { Knex } from 'knex'; +import { DB_MIGRATIONS_TABLE } from './tables'; + +export async function migrateBackendTasks(knex: Knex): Promise { + const migrationsDir = resolvePackagePath( + '@backstage/backend-defaults', + 'migrations/scheduler', + ); + + await knex.migrate.latest({ + directory: migrationsDir, + tableName: DB_MIGRATIONS_TABLE, + }); +} diff --git a/packages/backend-defaults/src/entrypoints/scheduler/database/tables.ts b/packages/backend-defaults/src/entrypoints/scheduler/database/tables.ts new file mode 100644 index 0000000000..63aad6e42a --- /dev/null +++ b/packages/backend-defaults/src/entrypoints/scheduler/database/tables.ts @@ -0,0 +1,27 @@ +/* + * Copyright 2021 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. + */ + +export const DB_MIGRATIONS_TABLE = 'backstage_backend_tasks__knex_migrations'; +export const DB_TASKS_TABLE = 'backstage_backend_tasks__tasks'; + +export type DbTasksRow = { + id: string; + settings_json: string; + next_run_start_at: Date; + current_run_ticket?: string; + current_run_started_at?: Date | string; + current_run_expires_at?: Date | string; +}; diff --git a/packages/backend-plugin-api/src/entrypoints/scheduler/index.ts b/packages/backend-defaults/src/entrypoints/scheduler/index.ts similarity index 68% rename from packages/backend-plugin-api/src/entrypoints/scheduler/index.ts rename to packages/backend-defaults/src/entrypoints/scheduler/index.ts index 8c7dbf65f6..77d4ea561d 100644 --- a/packages/backend-plugin-api/src/entrypoints/scheduler/index.ts +++ b/packages/backend-defaults/src/entrypoints/scheduler/index.ts @@ -14,13 +14,4 @@ * limitations under the License. */ -export { readTaskScheduleDefinitionFromConfig } from './readTaskScheduleDefinitionFromConfig'; -export type { - SchedulerService, - TaskDescriptor, - TaskFunction, - TaskInvocationDefinition, - TaskRunner, - TaskScheduleDefinition, - TaskScheduleDefinitionConfig, -} from './types'; +export { schedulerServiceFactory } from './schedulerServiceFactory'; diff --git a/packages/backend-defaults/src/entrypoints/scheduler/lib/LocalTaskWorker.test.ts b/packages/backend-defaults/src/entrypoints/scheduler/lib/LocalTaskWorker.test.ts new file mode 100644 index 0000000000..4fd4949478 --- /dev/null +++ b/packages/backend-defaults/src/entrypoints/scheduler/lib/LocalTaskWorker.test.ts @@ -0,0 +1,109 @@ +/* + * 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 { 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).toHaveBeenCalledTimes(0); + await new Promise(r => setTimeout(r, 100)); + expect(fn).toHaveBeenCalledTimes(0); + await new Promise(r => setTimeout(r, 200)); + expect(fn).toHaveBeenCalledTimes(1); + await new Promise(r => setTimeout(r, 200)); + expect(fn).toHaveBeenCalledTimes(2); + controller.abort(); + await new Promise(r => setTimeout(r, 200)); + expect(fn).toHaveBeenCalledTimes(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).toHaveBeenCalledTimes(0); + await new Promise(r => setTimeout(r, 100)); + expect(fn).toHaveBeenCalledTimes(0); + await new Promise(r => setTimeout(r, 200)); + expect(fn).toHaveBeenCalledTimes(1); + await new Promise(r => setTimeout(r, 1000)); + expect(fn).toHaveBeenCalledTimes(2); + controller.abort(); + await new Promise(r => setTimeout(r, 1000)); + expect(fn).toHaveBeenCalledTimes(2); + }); + + it('can trigger to abort wait', 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).toHaveBeenCalledTimes(0); + await new Promise(r => setTimeout(r, 100)); + expect(fn).toHaveBeenCalledTimes(0); + await new Promise(r => setTimeout(r, 200)); + expect(fn).toHaveBeenCalledTimes(1); + worker.trigger(); + await new Promise(r => setTimeout(r, 10)); + expect(fn).toHaveBeenCalledTimes(2); + controller.abort(); + }); +}); diff --git a/packages/backend-defaults/src/entrypoints/scheduler/lib/LocalTaskWorker.ts b/packages/backend-defaults/src/entrypoints/scheduler/lib/LocalTaskWorker.ts new file mode 100644 index 0000000000..3510a855e9 --- /dev/null +++ b/packages/backend-defaults/src/entrypoints/scheduler/lib/LocalTaskWorker.ts @@ -0,0 +1,155 @@ +/* + * 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 { LoggerService } from '@backstage/backend-plugin-api'; +import { SchedulerServiceTaskFunction } from '@backstage/backend-plugin-api'; +import { ConflictError } from '@backstage/errors'; +import { CronTime } from 'cron'; +import { DateTime, Duration } from 'luxon'; +import { TaskSettingsV2 } from './types'; +import { delegateAbortController, sleep } from './util'; + +/** + * Implements tasks that run locally without cross-host collaboration. + * + * @private + */ +export class LocalTaskWorker { + private abortWait: AbortController | undefined; + + constructor( + private readonly taskId: string, + private readonly fn: SchedulerServiceTaskFunction, + private readonly logger: LoggerService, + ) {} + + start(settings: TaskSettingsV2, options?: { signal?: AbortSignal }) { + this.logger.info( + `Task worker starting: ${this.taskId}, ${JSON.stringify(settings)}`, + ); + + (async () => { + let attemptNum = 1; + for (;;) { + try { + if (settings.initialDelayDuration) { + await this.sleep( + Duration.fromISO(settings.initialDelayDuration), + options?.signal, + ); + } + + while (!options?.signal?.aborted) { + const startTime = process.hrtime(); + await this.runOnce(settings, options?.signal); + const timeTaken = process.hrtime(startTime); + await this.waitUntilNext( + settings, + (timeTaken[0] + timeTaken[1] / 1e9) * 1000, + options?.signal, + ); + } + + this.logger.info(`Task worker finished: ${this.taskId}`); + attemptNum = 0; + break; + } catch (e) { + attemptNum += 1; + this.logger.warn( + `Task worker failed unexpectedly, attempt number ${attemptNum}, ${e}`, + ); + await sleep(Duration.fromObject({ seconds: 1 })); + } + } + })(); + } + + 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().toJSDate(); + 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), + )}`, + ); + + await this.sleep(Duration.fromMillis(dt), signal); + } + + private async sleep( + duration: Duration, + abortSignal?: AbortSignal, + ): Promise { + this.abortWait = delegateAbortController(abortSignal); + await sleep(duration, this.abortWait.signal); + this.abortWait.abort(); // cleans up resources + this.abortWait = undefined; + } +} diff --git a/packages/backend-defaults/src/entrypoints/scheduler/lib/PluginTaskSchedulerImpl.test.ts b/packages/backend-defaults/src/entrypoints/scheduler/lib/PluginTaskSchedulerImpl.test.ts new file mode 100644 index 0000000000..57faa6faf8 --- /dev/null +++ b/packages/backend-defaults/src/entrypoints/scheduler/lib/PluginTaskSchedulerImpl.test.ts @@ -0,0 +1,349 @@ +/* + * Copyright 2021 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 { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils'; +import { ConflictError, NotFoundError } from '@backstage/errors'; +import { Duration } from 'luxon'; +import { migrateBackendTasks } from '../database/migrateBackendTasks'; +import { + PluginTaskSchedulerImpl, + parseDuration, +} from './PluginTaskSchedulerImpl'; + +function defer() { + let resolve = () => {}; + const promise = new Promise(_resolve => { + resolve = _resolve; + }); + return { promise, resolve }; +} + +jest.setTimeout(60_000); + +describe('PluginTaskManagerImpl', () => { + const databases = TestDatabases.create({ + ids: ['POSTGRES_16', 'POSTGRES_12', 'SQLITE_3'], + }); + + beforeAll(async () => { + // Make sure all databases are running before mocking timers, in case of testcontainers + await Promise.all( + databases.eachSupportedId().map(([id]) => databases.init(id)), + ); + + jest.useFakeTimers(); + }, 60_000); + + async function init(databaseId: TestDatabaseId) { + const knex = await databases.init(databaseId); + await migrateBackendTasks(knex); + const manager = new PluginTaskSchedulerImpl( + async () => knex, + getVoidLogger(), + ); + return { knex, manager }; + } + + // This is just to test the wrapper code; most of the actual tests are in + // TaskWorker.test.ts + describe('scheduleTask with global scope', () => { + it.each(databases.eachSupportedId())( + 'can run the v1 happy path, %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 promise; + expect(fn).toHaveBeenCalledWith(expect.any(AbortSignal)); + }, + ); + + it.each(databases.eachSupportedId())( + 'can run the v2 happy path, %p', + async databaseId => { + const { manager } = await init(databaseId); + + const fn = jest.fn(); + const promise = new Promise(resolve => fn.mockImplementation(resolve)); + await manager.scheduleTask({ + id: 'task2', + timeout: Duration.fromMillis(5000), + frequency: { cron: '* * * * * *' }, + fn, + scope: 'global', + }); + + await promise; + expect(fn).toHaveBeenCalledWith(expect.any(AbortSignal)); + }, + ); + }); + + describe('triggerTask with global scope', () => { + it.each(databases.eachSupportedId())( + 'can manually trigger a task, %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.fromObject({ years: 1 }), + initialDelay: Duration.fromObject({ years: 1 }), + fn, + scope: 'global', + }); + + await manager.triggerTask('task1'); + jest.advanceTimersByTime(5000); + + await promise; + expect(fn).toHaveBeenCalledWith(expect.any(AbortSignal)); + }, + ); + + it.each(databases.eachSupportedId())( + 'cant trigger a non-existent task, %p', + async databaseId => { + const { manager } = await init(databaseId); + + const fn = jest.fn(); + await manager.scheduleTask({ + id: 'task1', + timeout: Duration.fromMillis(5000), + frequency: Duration.fromObject({ years: 1 }), + fn, + scope: 'global', + }); + + await expect(() => manager.triggerTask('task2')).rejects.toThrow( + NotFoundError, + ); + }, + ); + + it.each(databases.eachSupportedId())( + 'cant trigger a running task, %p', + async databaseId => { + const { manager } = await init(databaseId); + + const { promise, resolve } = defer(); + + await manager.scheduleTask({ + id: 'task1', + timeout: Duration.fromMillis(5000), + frequency: Duration.fromObject({ years: 1 }), + fn: async () => { + resolve(); + await new Promise(r => setTimeout(r, 20000)); + }, + scope: 'global', + }); + + await promise; + await expect(() => manager.triggerTask('task1')).rejects.toThrow( + ConflictError, + ); + }, + ); + }); + + // This is just to test the wrapper code; most of the actual tests are in + // TaskWorker.test.ts + describe('scheduleTask with local scope', () => { + it('can run the v1 happy path', async () => { + const { manager } = await init('SQLITE_3'); + + const fn = jest.fn(); + const promise = new Promise(resolve => fn.mockImplementation(resolve)); + await manager.scheduleTask({ + id: 'task1', + timeout: { milliseconds: 5000 }, + frequency: { milliseconds: 5000 }, + fn, + scope: 'local', + }); + + await promise; + expect(fn).toHaveBeenCalledWith(expect.any(AbortSignal)); + }, 60_000); + + it('can run the v2 happy path', async () => { + const { manager } = await init('SQLITE_3'); + + const fn = jest.fn(); + const promise = new Promise(resolve => fn.mockImplementation(resolve)); + await manager.scheduleTask({ + id: 'task2', + timeout: Duration.fromMillis(5000), + frequency: { cron: '* * * * * *' }, + fn, + scope: 'local', + }); + + await promise; + expect(fn).toHaveBeenCalledWith(expect.any(AbortSignal)); + }, 60_000); + }); + + describe('triggerTask with local scope', () => { + it('can manually trigger a task', async () => { + const { manager } = await init('SQLITE_3'); + + const fn = jest.fn(); + const promise = new Promise(resolve => fn.mockImplementation(resolve)); + await manager.scheduleTask({ + id: 'task1', + timeout: Duration.fromMillis(5000), + frequency: Duration.fromObject({ years: 1 }), + initialDelay: Duration.fromObject({ years: 1 }), + fn, + scope: 'local', + }); + + await manager.triggerTask('task1'); + jest.advanceTimersByTime(5000); + + await promise; + expect(fn).toHaveBeenCalledWith(expect.any(AbortSignal)); + }, 60_000); + + it('cant trigger a non-existent task', async () => { + const { manager } = await init('SQLITE_3'); + + const fn = jest.fn(); + await manager.scheduleTask({ + id: 'task1', + timeout: Duration.fromMillis(5000), + frequency: Duration.fromObject({ years: 1 }), + fn, + scope: 'local', + }); + + await expect(() => manager.triggerTask('task2')).rejects.toThrow( + NotFoundError, + ); + }, 60_000); + + it('cant trigger a running task', async () => { + const { manager } = await init('SQLITE_3'); + + const { promise, resolve } = defer(); + + await manager.scheduleTask({ + id: 'task1', + timeout: Duration.fromMillis(5000), + frequency: Duration.fromObject({ years: 1 }), + fn: async () => { + resolve(); + await new Promise(r => setTimeout(r, 20000)); + }, + scope: 'local', + }); + + await promise; + await expect(() => manager.triggerTask('task1')).rejects.toThrow( + ConflictError, + ); + }, 60_000); + }); + + // This is just to test the wrapper code; most of the actual tests are in + // TaskWorker.test.ts + describe('createScheduledTaskRunner', () => { + it.each(databases.eachSupportedId())( + 'can run the happy path, %p', + async databaseId => { + const { manager } = await init(databaseId); + + const fn = jest.fn(); + const promise = new Promise(resolve => fn.mockImplementation(resolve)); + await manager + .createScheduledTaskRunner({ + timeout: Duration.fromMillis(5000), + frequency: Duration.fromMillis(5000), + scope: 'global', + }) + .run({ + id: 'task1', + fn, + }); + + await promise; + expect(fn).toHaveBeenCalledWith(expect.any(AbortSignal)); + }, + ); + }); + + 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(); + + 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 expect(manager.getScheduledTasks()).resolves.toEqual([ + { + id: 'task1', + scope: 'global', + settings: expect.objectContaining({ cadence: 'PT5S' }), + }, + { + id: 'task2', + scope: 'local', + settings: expect.objectContaining({ cadence: 'PT5S' }), + }, + ]); + }, + ); + }); + + describe('parseDuration', () => { + it('should parse durations', () => { + expect(parseDuration({ milliseconds: 5000 })).toEqual('PT5S'); + expect(parseDuration(Duration.fromMillis(5000))).toEqual('PT5S'); + expect(parseDuration({ cron: '1 * * * *' })).toEqual('1 * * * *'); + }); + }); +}); diff --git a/packages/backend-defaults/src/entrypoints/scheduler/lib/PluginTaskSchedulerImpl.ts b/packages/backend-defaults/src/entrypoints/scheduler/lib/PluginTaskSchedulerImpl.ts new file mode 100644 index 0000000000..ed14d75db2 --- /dev/null +++ b/packages/backend-defaults/src/entrypoints/scheduler/lib/PluginTaskSchedulerImpl.ts @@ -0,0 +1,169 @@ +/* + * Copyright 2021 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 { LoggerService } from '@backstage/backend-plugin-api'; +import { + SchedulerServiceTaskDescriptor, + SchedulerServiceTaskFunction, + SchedulerServiceTaskInvocationDefinition, + SchedulerServiceTaskRunner, + SchedulerServiceTaskScheduleDefinition, +} from '@backstage/backend-plugin-api'; +import { Counter, Histogram, metrics } from '@opentelemetry/api'; +import { Knex } from 'knex'; +import { Duration } from 'luxon'; +import { LocalTaskWorker } from './LocalTaskWorker'; +import { TaskWorker } from './TaskWorker'; +import { PluginTaskScheduler, TaskSettingsV2 } from './types'; +import { validateId } from './util'; + +/** + * Implements the actual task management. + */ +export class PluginTaskSchedulerImpl implements PluginTaskScheduler { + private readonly localTasksById = new Map(); + private readonly allScheduledTasks: SchedulerServiceTaskDescriptor[] = []; + + private readonly counter: Counter; + private readonly duration: Histogram; + + constructor( + private readonly databaseFactory: () => Promise, + private readonly logger: LoggerService, + ) { + const meter = metrics.getMeter('default'); + this.counter = meter.createCounter('backend_tasks.task.runs.count', { + description: 'Total number of times a task has been run', + }); + this.duration = meter.createHistogram('backend_tasks.task.runs.duration', { + description: 'Histogram of task run durations', + unit: 'seconds', + }); + } + + async triggerTask(id: string): Promise { + const localTask = this.localTasksById.get(id); + if (localTask) { + localTask.trigger(); + return; + } + + const knex = await this.databaseFactory(); + await TaskWorker.trigger(knex, id); + } + + async scheduleTask( + task: SchedulerServiceTaskScheduleDefinition & + SchedulerServiceTaskInvocationDefinition, + ): Promise { + 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( + task.id, + this.wrapInMetrics(task.fn, { labels: { taskId: task.id, scope } }), + knex, + this.logger.child({ task: task.id }), + ); + await worker.start(settings, { signal: task.signal }); + } else { + const worker = new LocalTaskWorker( + task.id, + this.wrapInMetrics(task.fn, { labels: { taskId: task.id, scope } }), + this.logger.child({ task: task.id }), + ); + worker.start(settings, { signal: task.signal }); + this.localTasksById.set(task.id, worker); + } + + this.allScheduledTasks.push({ + id: task.id, + scope: scope, + settings: settings, + }); + } + + createScheduledTaskRunner( + schedule: SchedulerServiceTaskScheduleDefinition, + ): SchedulerServiceTaskRunner { + return { + run: async task => { + await this.scheduleTask({ ...task, ...schedule }); + }, + }; + } + + async getScheduledTasks(): Promise { + return this.allScheduledTasks; + } + + private wrapInMetrics( + fn: SchedulerServiceTaskFunction, + opts: { labels: Record }, + ): SchedulerServiceTaskFunction { + return async abort => { + const labels = { + ...opts.labels, + }; + this.counter.add(1, { ...labels, result: 'started' }); + + const startTime = process.hrtime(); + + try { + await fn(abort); + labels.result = 'completed'; + } catch (ex) { + labels.result = 'failed'; + throw ex; + } finally { + const delta = process.hrtime(startTime); + const endTime = delta[0] + delta[1] / 1e9; + this.counter.add(1, labels); + this.duration.record(endTime, labels); + } + }; + } +} + +export function parseDuration( + frequency: SchedulerServiceTaskScheduleDefinition['frequency'], +): string { + if ('cron' in frequency) { + return frequency.cron; + } + + const parsed = Duration.isDuration(frequency) + ? frequency + : Duration.fromObject(frequency); + + if (!parsed.isValid) { + throw new Error( + `Invalid duration, ${parsed.invalidReason}: ${parsed.invalidExplanation}`, + ); + } + + return parsed.toISO()!; +} diff --git a/packages/backend-defaults/src/entrypoints/scheduler/lib/PluginTaskSchedulerJanitor.test.ts b/packages/backend-defaults/src/entrypoints/scheduler/lib/PluginTaskSchedulerJanitor.test.ts new file mode 100644 index 0000000000..82133a9e54 --- /dev/null +++ b/packages/backend-defaults/src/entrypoints/scheduler/lib/PluginTaskSchedulerJanitor.test.ts @@ -0,0 +1,96 @@ +/* + * Copyright 2021 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 { TestDatabases } from '@backstage/backend-test-utils'; +import { Knex } from 'knex'; +import { Duration } from 'luxon'; +import waitForExpect from 'wait-for-expect'; +import { migrateBackendTasks } from '../database/migrateBackendTasks'; +import { DB_TASKS_TABLE, DbTasksRow } from '../database/tables'; +import { PluginTaskSchedulerJanitor } from './PluginTaskSchedulerJanitor'; +import { createTestScopedSignal } from './__testUtils__/createTestScopedSignal'; + +const insertTask = async (knex: Knex, task: DbTasksRow) => { + return knex(DB_TASKS_TABLE) + .insert(task) + .onConflict('id') + .merge(['settings_json']); +}; + +const getTask = async (knex: Knex): Promise => { + return (await knex(DB_TASKS_TABLE))[0]; +}; + +describe('PluginTaskSchedulerJanitor', () => { + const logger = getVoidLogger(); + const databases = TestDatabases.create({ + ids: [ + /* 'MYSQL_8' not supported yet */ + 'POSTGRES_16', + 'POSTGRES_12', + 'SQLITE_3', + 'MYSQL_8', + ], + }); + const testScopedSignal = createTestScopedSignal(); + + jest.setTimeout(60_000); + + beforeEach(() => { + jest.resetAllMocks(); + }); + + it.each(databases.eachSupportedId())( + 'Should update date if current_run_expires_at expires, %p', + async databaseId => { + const knex = await databases.init(databaseId); + await migrateBackendTasks(knex); + + const dateYesterday = new Date( + new Date().setDate(new Date().getDate() - 1), + ); + + await insertTask(knex, { + id: 'task1', + settings_json: '', + next_run_start_at: new Date('2023-03-01 00:00:00'), + current_run_ticket: 'ticket', + current_run_started_at: dateYesterday, + current_run_expires_at: dateYesterday, + }); + + const worker = new PluginTaskSchedulerJanitor({ + waitBetweenRuns: Duration.fromObject({ milliseconds: 20 }), + knex, + logger, + }); + + worker.start(testScopedSignal()); + + await waitForExpect(async () => { + await expect(getTask(knex)).resolves.toEqual( + expect.objectContaining({ + id: 'task1', + current_run_ticket: null, + current_run_started_at: null, + current_run_expires_at: null, + }), + ); + }); + }, + ); +}); diff --git a/packages/backend-defaults/src/entrypoints/scheduler/lib/PluginTaskSchedulerJanitor.ts b/packages/backend-defaults/src/entrypoints/scheduler/lib/PluginTaskSchedulerJanitor.ts new file mode 100644 index 0000000000..b0cd4572cf --- /dev/null +++ b/packages/backend-defaults/src/entrypoints/scheduler/lib/PluginTaskSchedulerJanitor.ts @@ -0,0 +1,96 @@ +/* + * Copyright 2021 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 { LoggerService } from '@backstage/backend-plugin-api'; +import { Knex } from 'knex'; +import { Duration } from 'luxon'; +import { DB_TASKS_TABLE, DbTasksRow } from '../database/tables'; +import { sleep } from './util'; + +/** + * Makes sure to auto-expire and clean up things that time out or for other + * reasons should not be left lingering. + */ +export class PluginTaskSchedulerJanitor { + private readonly knex: Knex; + private readonly waitBetweenRuns: Duration; + private readonly logger: LoggerService; + + constructor(options: { + knex: Knex; + waitBetweenRuns: Duration; + logger: LoggerService; + }) { + this.knex = options.knex; + this.waitBetweenRuns = options.waitBetweenRuns; + this.logger = options.logger; + } + + async start(abortSignal?: AbortSignal) { + while (!abortSignal?.aborted) { + try { + await this.runOnce(); + } catch (e) { + this.logger.warn(`Error while performing janitorial tasks, ${e}`); + } + + await sleep(this.waitBetweenRuns, abortSignal); + } + } + + private async runOnce() { + const dbNull = this.knex.raw('null'); + const configClient = this.knex.client.config.client; + + let tasks: Array<{ id: string }>; + if (configClient.includes('sqlite3') || configClient.includes('mysql')) { + tasks = await this.knex(DB_TASKS_TABLE) + .select('id') + .where('current_run_expires_at', '<', this.knex.fn.now()); + await this.knex(DB_TASKS_TABLE) + .whereIn( + 'id', + tasks.map(t => t.id), + ) + .update({ + current_run_ticket: dbNull, + current_run_started_at: dbNull, + current_run_expires_at: dbNull, + }); + } else { + tasks = await this.knex(DB_TASKS_TABLE) + .where('current_run_expires_at', '<', this.knex.fn.now()) + .update({ + current_run_ticket: dbNull, + current_run_started_at: dbNull, + current_run_expires_at: dbNull, + }) + .returning(['id']); + } + + // In rare cases, knex drivers may ignore "returning", and return the number + // of rows changed instead + if (typeof tasks === 'number') { + if (tasks > 0) { + this.logger.warn(`${tasks} tasks timed out and were lost`); + } + } else { + for (const { id } of tasks) { + this.logger.warn(`Task timed out and was lost: ${id}`); + } + } + } +} diff --git a/packages/backend-defaults/src/entrypoints/scheduler/lib/TaskScheduler.test.ts b/packages/backend-defaults/src/entrypoints/scheduler/lib/TaskScheduler.test.ts new file mode 100644 index 0000000000..d286eaee38 --- /dev/null +++ b/packages/backend-defaults/src/entrypoints/scheduler/lib/TaskScheduler.test.ts @@ -0,0 +1,84 @@ +/* + * Copyright 2021 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 { DatabaseManager, getVoidLogger } from '@backstage/backend-common'; +import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils'; +import { Duration } from 'luxon'; +import waitForExpect from 'wait-for-expect'; +import { TaskScheduler } from './TaskScheduler'; +import { createTestScopedSignal } from './__testUtils__/createTestScopedSignal'; + +jest.setTimeout(60_000); + +describe('TaskScheduler', () => { + const logger = getVoidLogger(); + const databases = TestDatabases.create(); + const testScopedSignal = createTestScopedSignal(); + + async function createDatabase( + databaseId: TestDatabaseId, + ): Promise { + const knex = await databases.init(databaseId); + const databaseManager: Partial = { + forPlugin: () => ({ + getClient: async () => knex, + }), + }; + return databaseManager as DatabaseManager; + } + + it.each(databases.eachSupportedId())( + 'can return a working v1 plugin impl, %p', + async databaseId => { + const database = await createDatabase(databaseId); + const manager = new TaskScheduler(database, logger).forPlugin('test'); + const fn = jest.fn(); + + await manager.scheduleTask({ + id: 'task1', + timeout: Duration.fromMillis(5000), + frequency: Duration.fromMillis(5000), + signal: testScopedSignal(), + fn, + }); + + await waitForExpect(() => { + expect(fn).toHaveBeenCalled(); + }); + }, + ); + + it.each(databases.eachSupportedId())( + 'can return a working v2 plugin impl, %p', + async databaseId => { + const database = await createDatabase(databaseId); + const manager = new TaskScheduler(database, logger).forPlugin('test'); + const fn = jest.fn(); + + await manager.scheduleTask({ + id: 'task2', + timeout: Duration.fromMillis(5000), + frequency: { cron: '* * * * * *' }, + signal: testScopedSignal(), + fn, + }); + + await waitForExpect(() => { + expect(fn).toHaveBeenCalled(); + }); + }, + ); +}); diff --git a/packages/backend-defaults/src/entrypoints/scheduler/lib/TaskScheduler.ts b/packages/backend-defaults/src/entrypoints/scheduler/lib/TaskScheduler.ts new file mode 100644 index 0000000000..0757d680ae --- /dev/null +++ b/packages/backend-defaults/src/entrypoints/scheduler/lib/TaskScheduler.ts @@ -0,0 +1,98 @@ +/* + * Copyright 2021 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 { + DatabaseManager, + getRootLogger, + LegacyRootDatabaseService, + PluginDatabaseManager, +} from '@backstage/backend-common'; +import { + LoggerService, + RootConfigService, +} from '@backstage/backend-plugin-api'; +import { once } from 'lodash'; +import { Duration } from 'luxon'; +import { migrateBackendTasks } from '../database/migrateBackendTasks'; +import { PluginTaskSchedulerImpl } from './PluginTaskSchedulerImpl'; +import { PluginTaskSchedulerJanitor } from './PluginTaskSchedulerJanitor'; +import { PluginTaskScheduler } from './types'; + +/** + * Deals with the scheduling of distributed tasks. + */ +export class TaskScheduler { + static fromConfig( + config: RootConfigService, + options?: { + databaseManager?: LegacyRootDatabaseService; + logger?: LoggerService; + }, + ): TaskScheduler { + const databaseManager = + options?.databaseManager ?? DatabaseManager.fromConfig(config); + const logger = (options?.logger || getRootLogger()).child({ + type: 'taskManager', + }); + return new TaskScheduler(databaseManager, logger); + } + + constructor( + private readonly databaseManager: LegacyRootDatabaseService, + private readonly logger: LoggerService, + ) {} + + /** + * Instantiates a task manager instance for the given plugin. + * + * @param pluginId - The unique ID of the plugin, for example "catalog" + * @returns A {@link PluginTaskScheduler} instance + */ + forPlugin(pluginId: string): PluginTaskScheduler { + return TaskScheduler.forPlugin({ + pluginId, + databaseManager: this.databaseManager.forPlugin(pluginId), + logger: this.logger, + }); + } + + static forPlugin(opts: { + pluginId: string; + databaseManager: PluginDatabaseManager; + logger: LoggerService; + }): PluginTaskScheduler { + const databaseFactory = once(async () => { + const knex = await opts.databaseManager.getClient(); + + if (!opts.databaseManager.migrations?.skip) { + await migrateBackendTasks(knex); + } + + if (process.env.NODE_ENV !== 'test') { + const janitor = new PluginTaskSchedulerJanitor({ + knex, + waitBetweenRuns: Duration.fromObject({ minutes: 1 }), + logger: opts.logger, + }); + janitor.start(); + } + + return knex; + }); + + return new PluginTaskSchedulerImpl(databaseFactory, opts.logger); + } +} diff --git a/packages/backend-defaults/src/entrypoints/scheduler/lib/TaskWorker.test.ts b/packages/backend-defaults/src/entrypoints/scheduler/lib/TaskWorker.test.ts new file mode 100644 index 0000000000..20d9b63859 --- /dev/null +++ b/packages/backend-defaults/src/entrypoints/scheduler/lib/TaskWorker.test.ts @@ -0,0 +1,507 @@ +/* + * Copyright 2021 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 { TestDatabases } from '@backstage/backend-test-utils'; +import { DateTime, Duration } from 'luxon'; +import waitForExpect from 'wait-for-expect'; +import { migrateBackendTasks } from '../database/migrateBackendTasks'; +import { DB_TASKS_TABLE, DbTasksRow } from '../database/tables'; +import { TaskWorker } from './TaskWorker'; +import { createTestScopedSignal } from './__testUtils__/createTestScopedSignal'; +import { TaskSettingsV2 } from './types'; + +jest.setTimeout(60_000); + +describe('TaskWorker', () => { + const logger = getVoidLogger(); + const databases = TestDatabases.create(); + const testScopedSignal = createTestScopedSignal(); + + beforeEach(() => { + jest.resetAllMocks(); + }); + + it.each(databases.eachSupportedId())( + 'goes through the expected states, %p', + async databaseId => { + const knex = await databases.init(databaseId); + await migrateBackendTasks(knex); + + const fn = jest.fn( + async () => new Promise(resolve => setTimeout(resolve, 50)), + ); + const settings: TaskSettingsV2 = { + version: 2, + cadence: '*/2 * * * * *', + initialDelayDuration: Duration.fromObject({ seconds: 1 }).toISO()!, + timeoutAfterDuration: Duration.fromObject({ minutes: 1 }).toISO()!, + }; + + const worker = new TaskWorker('task1', fn, knex, logger); + await worker.persistTask(settings); + + let row = (await knex(DB_TASKS_TABLE))[0]; + expect(row).toEqual( + expect.objectContaining({ + id: 'task1', + current_run_ticket: null, + current_run_started_at: null, + current_run_expires_at: null, + }), + ); + expect(JSON.parse(row.settings_json)).toEqual({ + version: 2, + cadence: '*/2 * * * * *', + initialDelayDuration: 'PT1S', + timeoutAfterDuration: 'PT1M', + }); + + await expect(worker.findReadyTask()).resolves.toEqual({ + result: 'not-ready-yet', + }); + + await waitForExpect(async () => { + await expect(worker.findReadyTask()).resolves.toEqual({ + result: 'ready', + settings, + }); + }); + + row = (await knex(DB_TASKS_TABLE))[0]; + expect(row).toEqual( + expect.objectContaining({ + id: 'task1', + current_run_ticket: null, + current_run_started_at: null, + current_run_expires_at: null, + }), + ); + + await expect(worker.tryClaimTask('ticket', settings)).resolves.toBe(true); + + row = (await knex(DB_TASKS_TABLE))[0]; + expect(row).toEqual( + expect.objectContaining({ + id: 'task1', + current_run_ticket: 'ticket', + current_run_started_at: expect.anything(), + current_run_expires_at: expect.anything(), + }), + ); + + await expect(worker.tryReleaseTask('ticket', settings)).resolves.toBe( + true, + ); + + row = (await knex(DB_TASKS_TABLE))[0]; + expect(row).toEqual( + expect.objectContaining({ + id: 'task1', + current_run_ticket: null, + current_run_started_at: null, + current_run_expires_at: null, + }), + ); + }, + ); + + it.each(databases.eachSupportedId())( + 'logs error when the task throws, %p', + async databaseId => { + const knex = await databases.init(databaseId); + await migrateBackendTasks(knex); + + jest.spyOn(logger, 'error'); + const fn = jest.fn().mockRejectedValue(new Error('failed')); + const settings: TaskSettingsV2 = { + version: 2, + initialDelayDuration: undefined, + cadence: '* * * * * *', + timeoutAfterDuration: Duration.fromMillis(60000).toISO()!, + }; + const checkFrequency = Duration.fromObject({ milliseconds: 100 }); + const worker = new TaskWorker('task1', fn, knex, logger, checkFrequency); + worker.start(settings, { signal: testScopedSignal() }); + + await waitForExpect(() => { + expect(logger.error).toHaveBeenCalled(); + }); + }, + ); + + it.each(databases.eachSupportedId())( + 'runs tasks more than once even when the task throws, %p', + async databaseId => { + const knex = await databases.init(databaseId); + await migrateBackendTasks(knex); + + const fn = jest.fn().mockRejectedValue(new Error('failed')); + const settings: TaskSettingsV2 = { + version: 2, + initialDelayDuration: undefined, + cadence: '* * * * * *', + timeoutAfterDuration: Duration.fromMillis(60000).toISO()!, + }; + const checkFrequency = Duration.fromObject({ milliseconds: 100 }); + const worker = new TaskWorker('task1', fn, knex, logger, checkFrequency); + worker.start(settings, { signal: testScopedSignal() }); + + await waitForExpect(() => { + expect(fn).toHaveBeenCalledTimes(3); + }); + }, + ); + + it.each(databases.eachSupportedId())( + 'does not clobber ticket lock when stolen, %p', + async databaseId => { + const knex = await databases.init(databaseId); + await migrateBackendTasks(knex); + + const fn = jest.fn( + async () => new Promise(resolve => setTimeout(resolve, 50)), + ); + const settings: TaskSettingsV2 = { + version: 2, + initialDelayDuration: undefined, + cadence: '* * * * * *', + timeoutAfterDuration: Duration.fromMillis(60000).toISO()!, + }; + + const worker = new TaskWorker('task1', fn, knex, logger); + await worker.persistTask(settings); + + await waitForExpect(async () => { + await expect(worker.findReadyTask()).resolves.toEqual({ + result: 'ready', + settings, + }); + }); + + await expect(worker.tryClaimTask('ticket', settings)).resolves.toBe(true); + + let row = (await knex(DB_TASKS_TABLE))[0]; + expect(row).toEqual( + expect.objectContaining({ + id: 'task1', + current_run_ticket: 'ticket', + current_run_started_at: expect.anything(), + current_run_expires_at: expect.anything(), + }), + ); + + await knex(DB_TASKS_TABLE) + .where('id', '=', 'task1') + .update({ current_run_ticket: 'stolen' }); + + await expect(worker.tryReleaseTask('ticket', settings)).resolves.toBe( + false, + ); + + row = (await knex(DB_TASKS_TABLE))[0]; + expect(row).toEqual( + expect.objectContaining({ + id: 'task1', + current_run_ticket: 'stolen', + current_run_started_at: expect.anything(), + current_run_expires_at: expect.anything(), + }), + ); + }, + ); + + it.each(databases.eachSupportedId())( + 'gracefully handles a disappeared task row, %p', + async databaseId => { + const knex = await databases.init(databaseId); + await migrateBackendTasks(knex); + + const fn = jest.fn(async () => {}); + const settings: TaskSettingsV2 = { + version: 2, + initialDelayDuration: undefined, + cadence: '* * * * * *', + timeoutAfterDuration: Duration.fromMillis(60000).toISO()!, + }; + + const worker1 = new TaskWorker('task1', fn, knex, logger); + await worker1.persistTask(settings); + await knex(DB_TASKS_TABLE).where('id', '=', 'task1').delete(); + await expect(worker1.findReadyTask()).resolves.toEqual({ + result: 'abort', + }); + + const worker2 = new TaskWorker('task2', fn, knex, logger); + await worker2.persistTask(settings); + + await waitForExpect(async () => { + await expect(worker2.findReadyTask()).resolves.toEqual({ + result: 'ready', + settings, + }); + }); + + await knex(DB_TASKS_TABLE).where('id', '=', 'task2').delete(); + await expect(worker2.tryClaimTask('ticket', settings)).resolves.toBe( + false, + ); + + const worker3 = new TaskWorker('task3', fn, knex, logger); + await worker3.persistTask(settings); + + await waitForExpect(async () => { + await expect(worker3.findReadyTask()).resolves.toEqual({ + result: 'ready', + settings, + }); + }); + + await expect(worker3.tryClaimTask('ticket', settings)).resolves.toBe( + true, + ); + await knex(DB_TASKS_TABLE).where('id', '=', 'task3').delete(); + await expect(worker3.tryReleaseTask('ticket', settings)).resolves.toBe( + false, + ); + }, + ); + + it.each(databases.eachSupportedId())( + 'respects initialDelayDuration per worker, %p', + async databaseId => { + const knex = await databases.init(databaseId); + await migrateBackendTasks(knex); + + const abortFirst = new AbortController(); + const settings: TaskSettingsV2 = { + version: 2, + initialDelayDuration: 'PT0.3S', + cadence: 'PT0.1S', + timeoutAfterDuration: 'PT10S', + }; + + // Start a single worker and make sure it waits and then goes to work + const fn1 = jest.fn(async () => {}); + const worker1 = new TaskWorker( + 'task1', + fn1, + knex, + logger, + Duration.fromMillis(10), + ); + await worker1.start(settings, { signal: abortFirst.signal }); + + expect(fn1).toHaveBeenCalledTimes(0); + await new Promise(resolve => setTimeout(resolve, 250)); + expect(fn1).toHaveBeenCalledTimes(0); + await new Promise(resolve => setTimeout(resolve, 100)); + expect(fn1.mock.calls.length).toBeGreaterThan(0); + + // Start a second worker and make sure it waits but the first worker still works along + const fn2 = jest.fn(); + const promise2 = new Promise(resolve => fn2.mockImplementation(resolve)); + const worker2 = new TaskWorker( + 'task1', + fn2, + knex, + logger, + Duration.fromMillis(10), + ); + await worker2.start(settings, { signal: testScopedSignal() }); + + // We eventually abort the first worker just to make sure that the second + // one for sure will get a go at running the task + setTimeout(() => abortFirst.abort(), 1000); + + const before = fn1.mock.calls.length; + await promise2; + expect(fn1.mock.calls.length).toBeGreaterThan(before); + + await knex.destroy(); + }, + ); + + it.each(databases.eachSupportedId())( + 'next_run_start_at is always the min between schedule changes from cron frequency, %p', + async databaseId => { + const knex = await databases.init(databaseId); + await migrateBackendTasks(knex); + + const fn = jest.fn( + async () => new Promise(resolve => setTimeout(resolve, 50)), + ); + const settings: TaskSettingsV2 = { + version: 2, + cadence: '*/15 * * * *', + initialDelayDuration: 'PT2M', + timeoutAfterDuration: 'PT1M', + }; + + const worker = new TaskWorker('task99', fn, knex, logger); + await worker.persistTask(settings); + const row1 = (await knex(DB_TASKS_TABLE))[0]; + + const settings2 = { + ...settings, + cadence: '*/2 * * * *', + initialDelayDuration: 'PT1M', + }; + await worker.persistTask(settings2); + const row2 = (await knex(DB_TASKS_TABLE))[0]; + + expect(row2.next_run_start_at).not.toStrictEqual(row1.next_run_start_at); + + const settings3 = { ...settings }; + await worker.persistTask(settings3); + const row3 = (await knex(DB_TASKS_TABLE))[0]; + + // The new timestamp can basically be 0 or a minute depending on how the + // initialDelayDuration falls right on a cron boundary. This kinda + // contrived check removes a test flakiness based on wall clock time. + expect( + Math.abs( + +new Date(row3.next_run_start_at) - +new Date(row2.next_run_start_at), + ), + ).toBeLessThanOrEqual(60_000); + + await knex.destroy(); + }, + ); + + it.each(databases.eachSupportedId())( + 'next_run_start_at is always the min between schedule changes when using human duration frequency, %p', + async databaseId => { + const knex = await databases.init(databaseId); + await migrateBackendTasks(knex); + + const fn = jest.fn( + async () => new Promise(resolve => setTimeout(resolve, 50)), + ); + + const initialSettings: TaskSettingsV2 = { + version: 2, + cadence: 'PT120M', + timeoutAfterDuration: 'PT1M', + }; + + const worker = new TaskWorker('task99', fn, knex, logger); + await worker.persistTask(initialSettings); + // replicate task running, sets next_run_start_at based on cadence + await worker.tryClaimTask('ticket', initialSettings); + await worker.tryReleaseTask('ticket', initialSettings); + + // grab initial row for comparisons later + const rowAfterClaimAndRelease = ( + await knex(DB_TASKS_TABLE) + )[0]; + + const settings: TaskSettingsV2 = { + ...initialSettings, + cadence: 'PT60M', + }; + await worker.persistTask(settings); + const row1 = (await knex(DB_TASKS_TABLE))[0]; + + const rowAfterClaimAndReleaseNextStartAt = DateTime.fromJSDate( + new Date(rowAfterClaimAndRelease.next_run_start_at), + ); + const row1NextStartAt = DateTime.fromJSDate( + new Date(row1.next_run_start_at), + ); + const now = DateTime.now(); + expect( + rowAfterClaimAndReleaseNextStartAt.diff(row1NextStartAt).as('minutes'), + ).toBeCloseTo(60, 1); // ensure that next start at is sooner than initial by one hour + expect(row1NextStartAt.diff(now).as('minutes')).toBeCloseTo(60, 1); // ensure that next start at is later than now by one hour + expect( + rowAfterClaimAndReleaseNextStartAt.diff(now).as('minutes'), + ).toBeCloseTo(120, 1); + + const settings2 = { + ...settings, + }; + await worker.persistTask(settings2); + const row2 = (await knex(DB_TASKS_TABLE))[0]; + + expect(row2.next_run_start_at).toStrictEqual(row1.next_run_start_at); + + await knex.destroy(); + }, + ); + + it.each(databases.eachSupportedId())( + 'next_run_start_at is always the min between schedule changes when using human duration frequency with initial start delay, %p', + async databaseId => { + const knex = await databases.init(databaseId); + await migrateBackendTasks(knex); + + const fn = jest.fn( + async () => new Promise(resolve => setTimeout(resolve, 50)), + ); + + const initialSettings: TaskSettingsV2 = { + version: 2, + cadence: 'PT120M', + initialDelayDuration: 'PT2M', + timeoutAfterDuration: 'PT1M', + }; + + const worker = new TaskWorker('task99', fn, knex, logger); + await worker.persistTask(initialSettings); + // replicate task running, sets next_run_start_at based on cadence + await worker.tryClaimTask('ticket', initialSettings); + await worker.tryReleaseTask('ticket', initialSettings); + + // grab initial row for comparisons later + const rowAfterClaimAndRelease = ( + await knex(DB_TASKS_TABLE) + )[0]; + + const settings: TaskSettingsV2 = { + ...initialSettings, + cadence: 'PT60M', + }; + await worker.persistTask(settings); + const row1 = (await knex(DB_TASKS_TABLE))[0]; + + const rowAfterClaimAndReleaseNextStartAt = DateTime.fromJSDate( + new Date(rowAfterClaimAndRelease.next_run_start_at), + ); + const row1NextStartAt = DateTime.fromJSDate( + new Date(row1.next_run_start_at), + ); + const now = DateTime.now(); + expect( + rowAfterClaimAndReleaseNextStartAt.diff(row1NextStartAt).as('minutes'), + ).toBeCloseTo(62, 1); // ensure that next start at is sooner than initial by one hour, plus the 2 minute delay (set my tryReleaseTask) + expect(row1NextStartAt.diff(now).as('minutes')).toBeCloseTo(60, 1); // ensure that next start at is later than now by one hour (2 minute delay doesn't take effect here) + expect( + rowAfterClaimAndReleaseNextStartAt.diff(now).as('minutes'), + ).toBeCloseTo(122, 1); // includes 2 minute start delay (which is persisted from tryReleaseTask) + + const settings2 = { + ...settings, + }; + await worker.persistTask(settings2); + const row2 = (await knex(DB_TASKS_TABLE))[0]; + + expect(row2.next_run_start_at).toStrictEqual(row1.next_run_start_at); + + await knex.destroy(); + }, + ); +}); diff --git a/packages/backend-defaults/src/entrypoints/scheduler/lib/TaskWorker.ts b/packages/backend-defaults/src/entrypoints/scheduler/lib/TaskWorker.ts new file mode 100644 index 0000000000..a1c4ec44fd --- /dev/null +++ b/packages/backend-defaults/src/entrypoints/scheduler/lib/TaskWorker.ts @@ -0,0 +1,373 @@ +/* + * Copyright 2021 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 { LoggerService } from '@backstage/backend-plugin-api'; +import { ConflictError, NotFoundError } from '@backstage/errors'; +import { CronTime } from 'cron'; +import { Knex } from 'knex'; +import { DateTime, Duration } from 'luxon'; +import { v4 as uuid } from 'uuid'; +import { DB_TASKS_TABLE, DbTasksRow } from '../database/tables'; +import { TaskSettingsV2, taskSettingsV2Schema } from './types'; +import { delegateAbortController, nowPlus, sleep } from './util'; +import { SchedulerServiceTaskFunction } from '@backstage/backend-plugin-api'; + +const DEFAULT_WORK_CHECK_FREQUENCY = Duration.fromObject({ seconds: 5 }); + +/** + * Implements tasks that run across worker hosts, with collaborative locking. + * + * @private + */ +export class TaskWorker { + constructor( + private readonly taskId: string, + private readonly fn: SchedulerServiceTaskFunction, + private readonly knex: Knex, + private readonly logger: LoggerService, + private readonly workCheckFrequency: Duration = DEFAULT_WORK_CHECK_FREQUENCY, + ) {} + + async start(settings: TaskSettingsV2, options?: { signal?: AbortSignal }) { + try { + await this.persistTask(settings); + } catch (e) { + throw new Error(`Failed to persist task, ${e}`); + } + + this.logger.info( + `Task worker starting: ${this.taskId}, ${JSON.stringify(settings)}`, + ); + + let workCheckFrequency = this.workCheckFrequency; + const isCron = !settings?.cadence.startsWith('P'); + if (!isCron) { + const cadence = Duration.fromISO(settings.cadence); + if (cadence < workCheckFrequency) { + workCheckFrequency = cadence; + } + } + + let attemptNum = 1; + (async () => { + for (;;) { + try { + if (settings.initialDelayDuration) { + await sleep( + Duration.fromISO(settings.initialDelayDuration), + options?.signal, + ); + } + + while (!options?.signal?.aborted) { + const runResult = await this.runOnce(options?.signal); + + if (runResult.result === 'abort') { + break; + } + + await sleep(workCheckFrequency, options?.signal); + } + + this.logger.info(`Task worker finished: ${this.taskId}`); + attemptNum = 0; + break; + } catch (e) { + attemptNum += 1; + this.logger.warn( + `Task worker failed unexpectedly, attempt number ${attemptNum}, ${e}`, + ); + await sleep(Duration.fromObject({ seconds: 1 })); + } + } + })(); + } + + 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 + */ + private async runOnce( + signal?: AbortSignal, + ): Promise< + | { result: 'not-ready-yet' } + | { result: 'abort' } + | { result: 'failed' } + | { result: 'completed' } + > { + const findResult = await this.findReadyTask(); + if ( + findResult.result === 'not-ready-yet' || + findResult.result === 'abort' + ) { + return findResult; + } + + const taskSettings = findResult.settings; + const ticket = uuid(); + + const claimed = await this.tryClaimTask(ticket, taskSettings); + if (!claimed) { + return { result: 'not-ready-yet' }; + } + + // 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(taskSettings.timeoutAfterDuration).as('milliseconds')); + + try { + await this.fn(taskAbortController.signal); + taskAbortController.abort(); // releases resources + } catch (e) { + this.logger.error(e); + await this.tryReleaseTask(ticket, taskSettings); + return { result: 'failed' }; + } finally { + clearTimeout(timeoutHandle); + } + + await this.tryReleaseTask(ticket, taskSettings); + return { result: 'completed' }; + } + + /** + * Perform the initial store of the task info + */ + async persistTask(settings: TaskSettingsV2) { + // Perform an initial parse to ensure that we will definitely be able to + // read it back again. + taskSettingsV2Schema.parse(settings); + + const isCron = !settings?.cadence.startsWith('P'); + + let startAt: Knex.Raw | undefined; + let nextStartAt: Knex.Raw | undefined; + if (settings.initialDelayDuration) { + startAt = nowPlus( + Duration.fromISO(settings.initialDelayDuration), + this.knex, + ); + } + + if (isCron) { + const time = new CronTime(settings.cadence) + .sendAt() + .minus({ seconds: 1 }) // immediately, if "* * * * * *" + .toUTC(); + + nextStartAt = this.nextRunAtRaw(time); + startAt ||= nextStartAt; + } else { + startAt ||= this.knex.fn.now(); + nextStartAt = nowPlus(Duration.fromISO(settings.cadence), this.knex); + } + + this.logger.debug(`task: ${this.taskId} configured to run at: ${startAt}`); + + // It's OK if the task already exists; if it does, just replace its + // settings with the new value and start the loop as usual. + const settingsJson = JSON.stringify(settings); + await this.knex(DB_TASKS_TABLE) + .insert({ + id: this.taskId, + settings_json: settingsJson, + next_run_start_at: startAt, + }) + .onConflict('id') + .merge( + this.knex.client.config.client.includes('mysql') + ? { + settings_json: settingsJson, + next_run_start_at: this.knex.raw( + `CASE WHEN ?? < ?? THEN ?? ELSE ?? END`, + [ + nextStartAt, + 'next_run_start_at', + nextStartAt, + 'next_run_start_at', + ], + ), + } + : { + settings_json: this.knex.ref('excluded.settings_json'), + next_run_start_at: this.knex.raw( + `CASE WHEN ?? < ?? THEN ?? ELSE ?? END`, + [ + nextStartAt, + `${DB_TASKS_TABLE}.next_run_start_at`, + nextStartAt, + `${DB_TASKS_TABLE}.next_run_start_at`, + ], + ), + }, + ); + } + + /** + * Check if the task is ready to run + */ + async findReadyTask(): Promise< + | { result: 'not-ready-yet' } + | { result: 'abort' } + | { result: 'ready'; settings: TaskSettingsV2 } + > { + const [row] = await this.knex(DB_TASKS_TABLE) + .where('id', '=', this.taskId) + .select({ + settingsJson: 'settings_json', + ready: this.knex.raw( + `CASE + WHEN next_run_start_at <= ? AND current_run_ticket IS NULL THEN TRUE + ELSE FALSE + END`, + [this.knex.fn.now()], + ), + }); + + if (!row) { + this.logger.info( + 'No longer able to find task; aborting and assuming that it has been unregistered or expired', + ); + return { result: 'abort' }; + } else if (!row.ready) { + return { result: 'not-ready-yet' }; + } + + try { + const obj = JSON.parse(row.settingsJson); + const settings = taskSettingsV2Schema.parse(obj); + return { result: 'ready', settings }; + } catch (e) { + this.logger.info( + `Task "${this.taskId}" is no longer able to parse task settings; aborting and assuming that a ` + + `newer version of the task has been issued and being handled by other workers, ${e}`, + ); + return { result: 'abort' }; + } + } + + /** + * Attempts to claim a task that's ready for execution, on this worker's + * behalf. We should not attempt to perform the work unless the claim really + * goes through. + * + * @param ticket - A globally unique string that changes for each invocation + * @param settings - The settings of the task to claim + * @returns True if it was successfully claimed + */ + async tryClaimTask( + ticket: string, + settings: TaskSettingsV2, + ): Promise { + const startedAt = this.knex.fn.now(); + const expiresAt = settings.timeoutAfterDuration + ? nowPlus(Duration.fromISO(settings.timeoutAfterDuration), this.knex) + : this.knex.raw('null'); + + const rows = await this.knex(DB_TASKS_TABLE) + .where('id', '=', this.taskId) + .whereNull('current_run_ticket') + .update({ + current_run_ticket: ticket, + current_run_started_at: startedAt, + current_run_expires_at: expiresAt, + }); + + return rows === 1; + } + + async tryReleaseTask( + ticket: string, + settings: TaskSettingsV2, + ): Promise { + const isCron = !settings?.cadence.startsWith('P'); + + let nextRun: Knex.Raw; + if (isCron) { + const time = new CronTime(settings.cadence).sendAt().toUTC(); + this.logger.debug(`task: ${this.taskId} will next occur around ${time}`); + + nextRun = this.nextRunAtRaw(time); + } else { + const dt = Duration.fromISO(settings.cadence).as('seconds'); + this.logger.debug( + `task: ${this.taskId} will next occur around ${DateTime.now().plus({ + seconds: dt, + })}`, + ); + + if (this.knex.client.config.client.includes('sqlite3')) { + nextRun = this.knex.raw( + `max(datetime(next_run_start_at, ?), datetime('now'))`, + [`+${dt} seconds`], + ); + } else if (this.knex.client.config.client.includes('mysql')) { + nextRun = this.knex.raw( + `greatest(next_run_start_at + interval ${dt} second, now())`, + ); + } else { + nextRun = this.knex.raw( + `greatest(next_run_start_at + interval '${dt} seconds', now())`, + ); + } + } + + const rows = await this.knex(DB_TASKS_TABLE) + .where('id', '=', this.taskId) + .where('current_run_ticket', '=', ticket) + .update({ + next_run_start_at: nextRun, + current_run_ticket: this.knex.raw('null'), + current_run_started_at: this.knex.raw('null'), + current_run_expires_at: this.knex.raw('null'), + }); + + return rows === 1; + } + + private nextRunAtRaw(time: DateTime): Knex.Raw { + if (this.knex.client.config.client.includes('sqlite3')) { + return this.knex.raw('datetime(?)', [time.toISO()]); + } else if (this.knex.client.config.client.includes('mysql')) { + return this.knex.raw(`?`, [time.toSQL({ includeOffset: false })]); + } + return this.knex.raw(`?`, [time.toISO()]); + } +} diff --git a/packages/backend-defaults/src/entrypoints/scheduler/lib/__testUtils__/createTestScopedSignal.ts b/packages/backend-defaults/src/entrypoints/scheduler/lib/__testUtils__/createTestScopedSignal.ts new file mode 100644 index 0000000000..6a497ea266 --- /dev/null +++ b/packages/backend-defaults/src/entrypoints/scheduler/lib/__testUtils__/createTestScopedSignal.ts @@ -0,0 +1,28 @@ +/* + * Copyright 2023 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. + */ + +export function createTestScopedSignal(): () => AbortSignal { + let testAbortController = new AbortController(); + + beforeEach(() => { + testAbortController = new AbortController(); + }); + afterEach(() => { + testAbortController.abort(); + }); + + return () => testAbortController.signal; +} diff --git a/packages/backend-defaults/src/entrypoints/scheduler/lib/types.ts b/packages/backend-defaults/src/entrypoints/scheduler/lib/types.ts new file mode 100644 index 0000000000..83f7b80073 --- /dev/null +++ b/packages/backend-defaults/src/entrypoints/scheduler/lib/types.ts @@ -0,0 +1,158 @@ +/* + * Copyright 2021 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 { + SchedulerServiceTaskDescriptor, + SchedulerServiceTaskInvocationDefinition, + SchedulerServiceTaskRunner, + SchedulerServiceTaskScheduleDefinition, +} from '@backstage/backend-plugin-api'; +import { CronTime } from 'cron'; +import { Duration } from 'luxon'; +import { z } from 'zod'; + +/** + * Deals with the scheduling of distributed tasks, for a given plugin. + */ +export interface PluginTaskScheduler { + /** + * Manually triggers a task by ID. + * + * If the task doesn't exist, a NotFoundError is thrown. If the task is + * currently running, a ConflictError is thrown. + * + * @param id - The task ID + */ + triggerTask(id: string): Promise; + + /** + * Schedules a task function for recurring runs. + * + * @remarks + * + * The `scope` task field controls whether to use coordinated exclusive + * invocation across workers, or to just coordinate within the current worker. + * + * This convenience method performs both the scheduling and invocation in one + * go. + * + * @param task - The task definition + */ + scheduleTask( + task: SchedulerServiceTaskScheduleDefinition & + SchedulerServiceTaskInvocationDefinition, + ): Promise; + + /** + * Creates a 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 + */ + createScheduledTaskRunner( + schedule: SchedulerServiceTaskScheduleDefinition, + ): SchedulerServiceTaskRunner; + + /** + * 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(): Promise; +} + +function isValidOptionalDurationString(d: string | undefined): boolean { + try { + return !d || Duration.fromISO(d).isValid; + } catch { + return false; + } +} + +function isValidCronFormat(c: string | undefined): boolean { + try { + if (!c) { + return false; + } + // parse cron format to ensure it's a valid format. + // eslint-disable-next-line no-new + new CronTime(c); + return true; + } catch { + return false; + } +} + +export const taskSettingsV1Schema = z.object({ + version: z.literal(1), + initialDelayDuration: z + .string() + .optional() + .refine(isValidOptionalDurationString, { + message: 'Invalid duration, expecting ISO Period', + }), + recurringAtMostEveryDuration: z + .string() + .refine(isValidOptionalDurationString, { + message: 'Invalid duration, expecting ISO Period', + }), + timeoutAfterDuration: z.string().refine(isValidOptionalDurationString, { + message: 'Invalid duration, expecting ISO Period', + }), +}); + +/** + * The properties that control a scheduled task (version 1). + */ +export type TaskSettingsV1 = z.infer; + +export const taskSettingsV2Schema = z.object({ + version: z.literal(2), + cadence: z + .string() + .refine(isValidCronFormat, { message: 'Invalid cron' }) + .or( + z.string().refine(isValidOptionalDurationString, { + message: 'Invalid duration, expecting ISO Period', + }), + ), + timeoutAfterDuration: z.string().refine(isValidOptionalDurationString, { + message: 'Invalid duration, expecting ISO Period', + }), + initialDelayDuration: z + .string() + .optional() + .refine(isValidOptionalDurationString, { + message: 'Invalid duration, expecting ISO Period', + }), +}); + +/** + * The properties that control a scheduled task (version 2). + */ +export type TaskSettingsV2 = z.infer; diff --git a/packages/backend-defaults/src/entrypoints/scheduler/lib/util.test.ts b/packages/backend-defaults/src/entrypoints/scheduler/lib/util.test.ts new file mode 100644 index 0000000000..e2536abb88 --- /dev/null +++ b/packages/backend-defaults/src/entrypoints/scheduler/lib/util.test.ts @@ -0,0 +1,113 @@ +/* + * Copyright 2021 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 knexFactory, { Knex } from 'knex'; +import { Duration } from 'luxon'; +import { delegateAbortController, nowPlus, sleep, validateId } from './util'; + +class KnexBuilder { + public build(client: string): Knex { + return knexFactory({ client, useNullAsDefault: true }); + } +} + +describe('util', () => { + describe('validateId', () => { + it.each(['a', 'a_b', 'ab123c_2', 'a!', 'A', 'a-b', 'a.b', '_a', 'a_'])( + 'accepts valid inputs, %p', + async input => { + expect(validateId(input)).toBeUndefined(); + }, + ); + + it.each(['', null, Symbol('a')])( + 'rejects invalid inputs, %p', + async input => { + expect(() => validateId(input as any)).toThrow(); + }, + ); + }); + + describe('sleep', () => { + it('finishes the wait as expected with no signal', async () => { + const ac = new AbortController(); + const start = Date.now(); + await sleep(Duration.fromObject({ seconds: 1 }), ac.signal); + expect(Date.now() - start).toBeGreaterThan(800); + }, 5_000); + + it('aborts properly on the signal', async () => { + const ac = new AbortController(); + const promise = sleep(Duration.fromObject({ seconds: 10 }), ac.signal); + ac.abort(); + await promise; + expect(true).toBe(true); + }, 1_000); + }); + + describe('delegateAbortController', () => { + it('inherits parent abort state', () => { + const parent = new AbortController(); + const child = delegateAbortController(parent.signal); + expect(parent.signal.aborted).toBe(false); + expect(child.signal.aborted).toBe(false); + parent.abort(); + expect(parent.signal.aborted).toBe(true); + expect(child.signal.aborted).toBe(true); + }); + + it('does not inherit from child to parent', () => { + const parent = new AbortController(); + const child = delegateAbortController(parent.signal); + expect(parent.signal.aborted).toBe(false); + expect(child.signal.aborted).toBe(false); + child.abort(); + expect(parent.signal.aborted).toBe(false); + expect(child.signal.aborted).toBe(true); + }); + }); + + describe('nowPlus', () => { + describe('without duration', () => { + const databases = [ + { client: 'sqlite3', expected: 'CURRENT_TIMESTAMP' }, + { client: 'mysql2', expected: 'CURRENT_TIMESTAMP' }, + { client: 'pg', expected: 'CURRENT_TIMESTAMP' }, + ]; + + it.each(databases)('for client $client', ({ client, expected }) => { + const knex = new KnexBuilder().build(client); + const result = nowPlus(undefined, knex); + + expect(result.toString()).toBe(expected); + }); + }); + describe('With duration', () => { + const databases = [ + { client: 'sqlite3', expected: "datetime('now', '20 seconds')" }, + { client: 'mysql2', expected: 'now() + interval 20 second' }, + { client: 'pg', expected: "now() + interval '20 seconds'" }, + ]; + it.each(databases)('for client $client', ({ client, expected }) => { + const duration = Duration.fromObject({ seconds: 20 }); + const knex = new KnexBuilder().build(client); + const result = nowPlus(duration, knex); + + expect(result.toString()).toBe(expected); + }); + }); + }); +}); diff --git a/packages/backend-defaults/src/entrypoints/scheduler/lib/util.ts b/packages/backend-defaults/src/entrypoints/scheduler/lib/util.ts new file mode 100644 index 0000000000..70d67a9fbe --- /dev/null +++ b/packages/backend-defaults/src/entrypoints/scheduler/lib/util.ts @@ -0,0 +1,111 @@ +/* + * Copyright 2021 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 { InputError } from '@backstage/errors'; +import { Knex } from 'knex'; +import { DateTime, Duration } from 'luxon'; + +// Keep the IDs compatible with e.g. Prometheus labels +export function validateId(id: string) { + if (typeof id !== 'string' || !id.trim()) { + throw new InputError(`${id} is not a valid ID, expected non-empty string`); + } +} + +export function dbTime(t: Date | string): DateTime { + if (typeof t === 'string') { + return DateTime.fromSQL(t); + } + return DateTime.fromJSDate(t); +} + +export function nowPlus(duration: Duration | undefined, knex: Knex) { + const seconds = duration?.as('seconds') ?? 0; + if (!seconds) { + return knex.fn.now(); + } + + if (knex.client.config.client.includes('sqlite3')) { + return knex.raw(`datetime('now', ?)`, [`${seconds} seconds`]); + } + + if (knex.client.config.client.includes('mysql')) { + return knex.raw(`now() + interval ${seconds} second`); + } + + return knex.raw(`now() + interval '${seconds} seconds'`); +} + +/** + * Sleep for the given duration, but return sooner if the abort signal + * triggers. + * + * @param duration - The amount of time to sleep, at most + * @param abortSignal - An optional abort signal that short circuits the wait + */ +export async function sleep( + duration: Duration, + abortSignal?: AbortSignal, +): Promise { + if (abortSignal?.aborted) { + return; + } + + await new Promise(resolve => { + let timeoutHandle: NodeJS.Timeout | undefined = undefined; + + const done = () => { + if (timeoutHandle) { + clearTimeout(timeoutHandle); + } + abortSignal?.removeEventListener('abort', done); + resolve(); + }; + + timeoutHandle = setTimeout(done, duration.as('milliseconds')); + abortSignal?.addEventListener('abort', done); + }); +} + +/** + * Creates a new AbortController that, in addition to working as a regular + * standalone controller, also gets aborted if the given parent signal + * reaches aborted state. + * + * @param parent - The "parent" signal that can trigger the delegate + */ +export function delegateAbortController(parent?: AbortSignal): AbortController { + const delegate = new AbortController(); + + if (parent) { + if (parent.aborted) { + delegate.abort(); + } else { + const onParentAborted = () => { + delegate.abort(); + }; + + const onChildAborted = () => { + parent.removeEventListener('abort', onParentAborted); + }; + + parent.addEventListener('abort', onParentAborted, { once: true }); + delegate.signal.addEventListener('abort', onChildAborted, { once: true }); + } + } + + return delegate; +} diff --git a/packages/backend-defaults/src/entrypoints/scheduler/schedulerServiceFactory.test.ts b/packages/backend-defaults/src/entrypoints/scheduler/schedulerServiceFactory.test.ts new file mode 100644 index 0000000000..7b889b3a0e --- /dev/null +++ b/packages/backend-defaults/src/entrypoints/scheduler/schedulerServiceFactory.test.ts @@ -0,0 +1,46 @@ +/* + * Copyright 2023 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 { coreServices } from '@backstage/backend-plugin-api'; +import { ServiceFactoryTester } from '@backstage/backend-test-utils'; +import { schedulerServiceFactory } from './schedulerServiceFactory'; + +describe('schedulerFactory', () => { + it('creates sidecar database features', async () => { + const tester = ServiceFactoryTester.from(schedulerServiceFactory()); + + const scheduler = await tester.get(); + await scheduler.scheduleTask({ + id: 'task1', + timeout: { seconds: 1 }, + frequency: { seconds: 1 }, + fn: async () => {}, + }); + + const database = await tester.getService(coreServices.database); + + const client = await database.getClient(); + await expect( + client.from('backstage_backend_tasks__tasks').count(), + ).resolves.toEqual([{ 'count(*)': 1 }]); + await expect( + client.from('backstage_backend_tasks__knex_migrations').count(), + ).resolves.toEqual([{ 'count(*)': expect.any(Number) }]); + await expect( + client.from('backstage_backend_tasks__knex_migrations_lock').count(), + ).resolves.toEqual([{ 'count(*)': expect.any(Number) }]); + }); +}); diff --git a/packages/backend-defaults/src/entrypoints/scheduler/schedulerServiceFactory.ts b/packages/backend-defaults/src/entrypoints/scheduler/schedulerServiceFactory.ts new file mode 100644 index 0000000000..55c0d58e7b --- /dev/null +++ b/packages/backend-defaults/src/entrypoints/scheduler/schedulerServiceFactory.ts @@ -0,0 +1,42 @@ +/* + * 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 { + coreServices, + createServiceFactory, +} from '@backstage/backend-plugin-api'; +import { TaskScheduler } from './lib/TaskScheduler'; + +/** + * The default service factory for {@link @backstage/backend-plugin-api#coreServices.scheduler}. + * + * @public + */ +export const schedulerServiceFactory = createServiceFactory({ + service: coreServices.scheduler, + deps: { + plugin: coreServices.pluginMetadata, + database: coreServices.database, + logger: coreServices.logger, + }, + async factory({ plugin, database, logger }) { + return TaskScheduler.forPlugin({ + pluginId: plugin.getId(), + databaseManager: database, + logger, + }); + }, +}); diff --git a/packages/backend-defaults/src/setupTests.ts b/packages/backend-defaults/src/setupTests.ts new file mode 100644 index 0000000000..76619a2542 --- /dev/null +++ b/packages/backend-defaults/src/setupTests.ts @@ -0,0 +1,25 @@ +/* + * Copyright 2020 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 { TestDatabases } from '@backstage/backend-test-utils'; +import { Settings } from 'luxon'; + +// TS still thinks that methods can return null / placeholders, but we still want to throw as soon as possible when things go wrong +Settings.throwOnInvalid = true; + +TestDatabases.setDefaults({ + ids: ['MYSQL_8', 'POSTGRES_16', 'POSTGRES_12', 'SQLITE_3'], +}); diff --git a/packages/backend-plugin-api/api-report-scheduler.md b/packages/backend-plugin-api/api-report-scheduler.md deleted file mode 100644 index 480c716025..0000000000 --- a/packages/backend-plugin-api/api-report-scheduler.md +++ /dev/null @@ -1,79 +0,0 @@ -## API Report File for "@backstage/backend-plugin-api" - -> Do not edit this file. It is a report generated by [API Extractor](https://api-extractor.com/). - -```ts -import { Config } from '@backstage/config'; -import { Duration } from 'luxon'; -import { HumanDuration } from '@backstage/types'; -import { JsonObject } from '@backstage/types'; - -// @public -export function readTaskScheduleDefinitionFromConfig( - config: Config, -): TaskScheduleDefinition; - -// @public -export interface SchedulerService { - createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner; - getScheduledTasks(): Promise; - scheduleTask( - task: TaskScheduleDefinition & TaskInvocationDefinition, - ): Promise; - triggerTask(id: string): Promise; -} - -// @public -export type TaskDescriptor = { - id: string; - scope: 'global' | 'local'; - settings: { - version: number; - } & JsonObject; -}; - -// @public -export type TaskFunction = - | ((abortSignal: AbortSignal) => void | Promise) - | (() => void | Promise); - -// @public -export interface TaskInvocationDefinition { - fn: TaskFunction; - id: string; - signal?: AbortSignal; -} - -// @public -export interface TaskRunner { - run(task: TaskInvocationDefinition): Promise; -} - -// @public -export interface TaskScheduleDefinition { - frequency: - | { - cron: string; - } - | Duration - | HumanDuration; - initialDelay?: Duration | HumanDuration; - scope?: 'global' | 'local'; - timeout: Duration | HumanDuration; -} - -// @public -export interface TaskScheduleDefinitionConfig { - frequency: - | { - cron: string; - } - | string - | HumanDuration; - initialDelay?: string | HumanDuration; - scope?: 'global' | 'local'; - timeout: string | HumanDuration; -} - -// (No @packageDocumentation comment for this package) -``` diff --git a/packages/backend-plugin-api/api-report.md b/packages/backend-plugin-api/api-report.md index 4e03da15b3..1ba2de768a 100644 --- a/packages/backend-plugin-api/api-report.md +++ b/packages/backend-plugin-api/api-report.md @@ -8,7 +8,9 @@ import { AuthorizePermissionRequest } from '@backstage/plugin-permission-common'; import { AuthorizePermissionResponse } from '@backstage/plugin-permission-common'; import { Config } from '@backstage/config'; +import { Duration } from 'luxon'; import { Handler } from 'express'; +import { HumanDuration } from '@backstage/types'; import { IdentityApi } from '@backstage/plugin-auth-node'; import { isChildPath } from '@backstage/cli-common'; import { JsonObject } from '@backstage/types'; @@ -20,7 +22,6 @@ import { QueryPermissionResponse } from '@backstage/plugin-permission-common'; import { Readable } from 'stream'; import { Request as Request_2 } from 'express'; import { Response as Response_2 } from 'express'; -import { SchedulerService as SchedulerService_2 } from '@backstage/backend-plugin-api/scheduler'; // @public (undocumented) export interface AuthService { @@ -202,7 +203,7 @@ export namespace coreServices { const rootHttpRouter: ServiceRef; const rootLifecycle: ServiceRef; const rootLogger: ServiceRef; - const scheduler: ServiceRef; + const scheduler: ServiceRef; const tokenManager: ServiceRef; const urlReader: ServiceRef; const identity: ServiceRef; @@ -450,6 +451,11 @@ export interface PluginServiceFactoryConfig< service: ServiceRef; } +// @public +export function readSchedulerServiceTaskScheduleDefinitionFromConfig( + config: Config, +): SchedulerServiceTaskScheduleDefinition; + // @public export type ReadTreeOptions = { filter?( @@ -536,8 +542,70 @@ export interface RootServiceFactoryConfig< service: ServiceRef; } -// @public @deprecated (undocumented) -export type SchedulerService = SchedulerService_2; +// @public +export interface SchedulerService { + createScheduledTaskRunner( + schedule: SchedulerServiceTaskScheduleDefinition, + ): SchedulerServiceTaskRunner; + getScheduledTasks(): Promise; + scheduleTask( + task: SchedulerServiceTaskScheduleDefinition & + SchedulerServiceTaskInvocationDefinition, + ): Promise; + triggerTask(id: string): Promise; +} + +// @public +export type SchedulerServiceTaskDescriptor = { + id: string; + scope: 'global' | 'local'; + settings: { + version: number; + } & JsonObject; +}; + +// @public +export type SchedulerServiceTaskFunction = + | ((abortSignal: AbortSignal) => void | Promise) + | (() => void | Promise); + +// @public +export interface SchedulerServiceTaskInvocationDefinition { + fn: SchedulerServiceTaskFunction; + id: string; + signal?: AbortSignal; +} + +// @public +export interface SchedulerServiceTaskRunner { + run(task: SchedulerServiceTaskInvocationDefinition): Promise; +} + +// @public +export interface SchedulerServiceTaskScheduleDefinition { + frequency: + | { + cron: string; + } + | Duration + | HumanDuration; + initialDelay?: Duration | HumanDuration; + scope?: 'global' | 'local'; + timeout: Duration | HumanDuration; +} + +// @public +export interface SchedulerServiceTaskScheduleDefinitionConfig { + frequency: + | { + cron: string; + } + | string + | HumanDuration; + initialDelay?: string | HumanDuration; + scope?: 'global' | 'local'; + timeout: string | HumanDuration; +} // @public export type SearchOptions = { diff --git a/packages/backend-plugin-api/package.json b/packages/backend-plugin-api/package.json index 3ffbb5b967..73668bad35 100644 --- a/packages/backend-plugin-api/package.json +++ b/packages/backend-plugin-api/package.json @@ -20,7 +20,6 @@ "license": "Apache-2.0", "exports": { ".": "./src/index.ts", - "./scheduler": "./src/entrypoints/scheduler/index.ts", "./alpha": "./src/alpha.ts", "./testUtils": "./src/testUtils.ts", "./package.json": "./package.json" @@ -29,9 +28,6 @@ "types": "src/index.ts", "typesVersions": { "*": { - "scheduler": [ - "src/entrypoints/scheduler/index.ts" - ], "alpha": [ "src/alpha.ts" ], diff --git a/packages/backend-plugin-api/src/entrypoints/scheduler/readTaskScheduleDefinitionFromConfig.ts b/packages/backend-plugin-api/src/entrypoints/scheduler/readTaskScheduleDefinitionFromConfig.ts deleted file mode 100644 index 8053936263..0000000000 --- a/packages/backend-plugin-api/src/entrypoints/scheduler/readTaskScheduleDefinitionFromConfig.ts +++ /dev/null @@ -1,78 +0,0 @@ -/* - * 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 { Config, readDurationFromConfig } from '@backstage/config'; -import { HumanDuration } from '@backstage/types'; -import { TaskScheduleDefinition } from './types'; -import { Duration } from 'luxon'; - -function readDuration(config: Config, key: string): Duration | HumanDuration { - if (typeof config.get(key) === 'string') { - const value = config.getString(key); - const duration = Duration.fromISO(value); - if (!duration.isValid) { - throw new Error(`Invalid duration: ${value}`); - } - return duration; - } - - return readDurationFromConfig(config, { key }); -} - -function readCronOrDuration( - config: Config, - key: string, -): { cron: string } | Duration | HumanDuration { - const value = config.get(key); - if (typeof value === 'object' && (value as { cron?: string }).cron) { - return value as { cron: string }; - } - - return readDuration(config, key); -} - -/** - * Reads a TaskScheduleDefinition from a Config. - * Expects the config not to be the root config, - * but the config for the definition. - * - * @param config - config for a TaskScheduleDefinition. - * @public - */ -export function readTaskScheduleDefinitionFromConfig( - config: Config, -): TaskScheduleDefinition { - const frequency = readCronOrDuration(config, 'frequency'); - const timeout = readDuration(config, 'timeout'); - - const initialDelay = config.has('initialDelay') - ? readDuration(config, 'initialDelay') - : undefined; - - const scope = config.getOptionalString('scope'); - if (scope && !['global', 'local'].includes(scope)) { - throw new Error( - `Only "global" or "local" are allowed for TaskScheduleDefinition.scope, but got: ${scope}`, - ); - } - - return { - frequency, - timeout, - initialDelay, - scope: scope as 'global' | 'local' | undefined, - }; -} diff --git a/packages/backend-plugin-api/src/index.ts b/packages/backend-plugin-api/src/index.ts index 342b33dd2e..c2e2a13a89 100644 --- a/packages/backend-plugin-api/src/index.ts +++ b/packages/backend-plugin-api/src/index.ts @@ -24,4 +24,3 @@ export * from './services'; export type { BackendFeature } from './types'; export * from './paths'; export * from './wiring'; -export * from './deprecated'; diff --git a/packages/backend-plugin-api/src/entrypoints/scheduler/readTaskScheduleDefinitionFromConfig.test.ts b/packages/backend-plugin-api/src/services/definitions/SchedulerService.test.ts similarity index 76% rename from packages/backend-plugin-api/src/entrypoints/scheduler/readTaskScheduleDefinitionFromConfig.test.ts rename to packages/backend-plugin-api/src/services/definitions/SchedulerService.test.ts index c52d59b016..7b20487af4 100644 --- a/packages/backend-plugin-api/src/entrypoints/scheduler/readTaskScheduleDefinitionFromConfig.test.ts +++ b/packages/backend-plugin-api/src/services/definitions/SchedulerService.test.ts @@ -17,9 +17,9 @@ import { ConfigReader } from '@backstage/config'; import { HumanDuration } from '@backstage/types'; import { Duration } from 'luxon'; -import { readTaskScheduleDefinitionFromConfig } from './readTaskScheduleDefinitionFromConfig'; +import { readSchedulerServiceTaskScheduleDefinitionFromConfig } from './SchedulerService'; -describe('readTaskScheduleDefinitionFromConfig', () => { +describe('readSchedulerServiceTaskScheduleDefinitionFromConfig', () => { it('all valid values', () => { const config = new ConfigReader({ frequency: { @@ -32,7 +32,7 @@ describe('readTaskScheduleDefinitionFromConfig', () => { scope: 'global', }); - const result = readTaskScheduleDefinitionFromConfig(config); + const result = readSchedulerServiceTaskScheduleDefinitionFromConfig(config); expect((result.frequency as { cron: string }).cron).toBe('0 30 * * * *'); expect(result.timeout).toEqual(Duration.fromISO('PT3M')); @@ -48,7 +48,7 @@ describe('readTaskScheduleDefinitionFromConfig', () => { timeout: 'PT3M', }); - const result = readTaskScheduleDefinitionFromConfig(config); + const result = readSchedulerServiceTaskScheduleDefinitionFromConfig(config); expect((result.frequency as { cron: string }).cron).toBe('0 30 * * * *'); expect(result.timeout).toEqual(Duration.fromISO('PT3M')); @@ -61,9 +61,9 @@ describe('readTaskScheduleDefinitionFromConfig', () => { timeout: 'PT3M', }); - expect(() => readTaskScheduleDefinitionFromConfig(config)).toThrow( - "Missing required config value at 'frequency'", - ); + expect(() => + readSchedulerServiceTaskScheduleDefinitionFromConfig(config), + ).toThrow("Missing required config value at 'frequency'"); }); it('fail without required timeout', () => { @@ -71,9 +71,9 @@ describe('readTaskScheduleDefinitionFromConfig', () => { frequency: 'PT30M', }); - expect(() => readTaskScheduleDefinitionFromConfig(config)).toThrow( - "Missing required config value at 'timeout'", - ); + expect(() => + readSchedulerServiceTaskScheduleDefinitionFromConfig(config), + ).toThrow("Missing required config value at 'timeout'"); }); it('invalid frequency key', () => { @@ -84,7 +84,9 @@ describe('readTaskScheduleDefinitionFromConfig', () => { timeout: 'PT3M', }); - expect(() => readTaskScheduleDefinitionFromConfig(config)).toThrow( + expect(() => + readSchedulerServiceTaskScheduleDefinitionFromConfig(config), + ).toThrow( "Failed to read duration from config at 'frequency', Error: Needs one or more of 'years', 'months', 'weeks', 'days', 'hours', 'minutes', 'seconds', 'milliseconds'", ); }); @@ -97,7 +99,9 @@ describe('readTaskScheduleDefinitionFromConfig', () => { timeout: 'PT3M', }); - expect(() => readTaskScheduleDefinitionFromConfig(config)).toThrow( + expect(() => + readSchedulerServiceTaskScheduleDefinitionFromConfig(config), + ).toThrow( "Failed to read duration from config, Error: Unable to convert config value for key 'frequency.minutes' in 'mock-config' to a number", ); }); @@ -111,7 +115,9 @@ describe('readTaskScheduleDefinitionFromConfig', () => { timeout: 'PT3M', }); - expect(() => readTaskScheduleDefinitionFromConfig(config)).toThrow( + expect(() => + readSchedulerServiceTaskScheduleDefinitionFromConfig(config), + ).toThrow( "Failed to read duration from config at 'frequency', Error: Unknown property 'invalid'; expected one or more of 'years', 'months', 'weeks', 'days', 'hours', 'minutes', 'seconds', 'milliseconds'", ); }); @@ -125,7 +131,9 @@ describe('readTaskScheduleDefinitionFromConfig', () => { scope: 'invalid', }); - expect(() => readTaskScheduleDefinitionFromConfig(config)).toThrow( + expect(() => + readSchedulerServiceTaskScheduleDefinitionFromConfig(config), + ).toThrow( 'Only "global" or "local" are allowed for TaskScheduleDefinition.scope, but got: invalid', ); }); diff --git a/packages/backend-plugin-api/src/entrypoints/scheduler/types.ts b/packages/backend-plugin-api/src/services/definitions/SchedulerService.ts similarity index 81% rename from packages/backend-plugin-api/src/entrypoints/scheduler/types.ts rename to packages/backend-plugin-api/src/services/definitions/SchedulerService.ts index 5b14e52fcb..ad0fc1cf3a 100644 --- a/packages/backend-plugin-api/src/entrypoints/scheduler/types.ts +++ b/packages/backend-plugin-api/src/services/definitions/SchedulerService.ts @@ -14,6 +14,7 @@ * limitations under the License. */ +import { Config, readDurationFromConfig } from '@backstage/config'; import { HumanDuration, JsonObject } from '@backstage/types'; import { Duration } from 'luxon'; @@ -25,7 +26,7 @@ import { Duration } from 'luxon'; * * @public */ -export type TaskFunction = +export type SchedulerServiceTaskFunction = | ((abortSignal: AbortSignal) => void | Promise) | (() => void | Promise); @@ -34,7 +35,7 @@ export type TaskFunction = * * @public */ -export type TaskDescriptor = { +export type SchedulerServiceTaskDescriptor = { /** * The unique identifier of the task. */ @@ -56,7 +57,7 @@ export type TaskDescriptor = { * * @public */ -export interface TaskScheduleDefinition { +export interface SchedulerServiceTaskScheduleDefinition { /** * How often you want the task to run. The system does its best to avoid * overlapping invocations. @@ -148,12 +149,12 @@ export interface TaskScheduleDefinition { } /** - * Config options for {@link TaskScheduleDefinition} + * Config options for {@link SchedulerServiceTaskScheduleDefinition} * that control the scheduling of a task. * * @public */ -export interface TaskScheduleDefinitionConfig { +export interface SchedulerServiceTaskScheduleDefinitionConfig { /** * How often you want the task to run. The system does its best to avoid * overlapping invocations. @@ -249,7 +250,7 @@ export interface TaskScheduleDefinitionConfig { * * @public */ -export interface TaskInvocationDefinition { +export interface SchedulerServiceTaskInvocationDefinition { /** * A unique ID (within the scope of the plugin) for the task. */ @@ -258,7 +259,7 @@ export interface TaskInvocationDefinition { /** * The actual task function to be invoked regularly. */ - fn: TaskFunction; + fn: SchedulerServiceTaskFunction; /** * An abort signal that, when triggered, will stop the recurring execution of @@ -272,13 +273,13 @@ export interface TaskInvocationDefinition { * * @public */ -export interface TaskRunner { +export interface SchedulerServiceTaskRunner { /** * Takes the schedule and executes an actual task using it. * * @param task - The actual runtime properties of the task */ - run(task: TaskInvocationDefinition): Promise; + run(task: SchedulerServiceTaskInvocationDefinition): Promise; } /** @@ -311,7 +312,8 @@ export interface SchedulerService { * @param task - The task definition */ scheduleTask( - task: TaskScheduleDefinition & TaskInvocationDefinition, + task: SchedulerServiceTaskScheduleDefinition & + SchedulerServiceTaskInvocationDefinition, ): Promise; /** @@ -326,7 +328,9 @@ export interface SchedulerService { * * @param schedule - The task schedule */ - createScheduledTaskRunner(schedule: TaskScheduleDefinition): TaskRunner; + createScheduledTaskRunner( + schedule: SchedulerServiceTaskScheduleDefinition, + ): SchedulerServiceTaskRunner; /** * Returns all scheduled tasks registered to this scheduler. @@ -339,5 +343,62 @@ export interface SchedulerService { * * @returns Scheduled tasks */ - getScheduledTasks(): Promise; + getScheduledTasks(): Promise; +} + +function readDuration(config: Config, key: string): Duration | HumanDuration { + if (typeof config.get(key) === 'string') { + const value = config.getString(key); + const duration = Duration.fromISO(value); + if (!duration.isValid) { + throw new Error(`Invalid duration: ${value}`); + } + return duration; + } + + return readDurationFromConfig(config, { key }); +} + +function readCronOrDuration( + config: Config, + key: string, +): { cron: string } | Duration | HumanDuration { + const value = config.get(key); + if (typeof value === 'object' && (value as { cron?: string }).cron) { + return value as { cron: string }; + } + + return readDuration(config, key); +} + +/** + * Reads a {@link SchedulerServiceTaskScheduleDefinition} from config. Expects + * the config not to be the root config, but the config for the definition. + * + * @param config - config for a TaskScheduleDefinition. + * @public + */ +export function readSchedulerServiceTaskScheduleDefinitionFromConfig( + config: Config, +): SchedulerServiceTaskScheduleDefinition { + const frequency = readCronOrDuration(config, 'frequency'); + const timeout = readDuration(config, 'timeout'); + + const initialDelay = config.has('initialDelay') + ? readDuration(config, 'initialDelay') + : undefined; + + const scope = config.getOptionalString('scope'); + if (scope && !['global', 'local'].includes(scope)) { + throw new Error( + `Only "global" or "local" are allowed for TaskScheduleDefinition.scope, but got: ${scope}`, + ); + } + + return { + frequency, + timeout, + initialDelay, + scope: scope as 'global' | 'local' | undefined, + }; } diff --git a/packages/backend-plugin-api/src/services/definitions/coreServices.ts b/packages/backend-plugin-api/src/services/definitions/coreServices.ts index 09f313af57..c760afc7b4 100644 --- a/packages/backend-plugin-api/src/services/definitions/coreServices.ts +++ b/packages/backend-plugin-api/src/services/definitions/coreServices.ts @@ -165,7 +165,7 @@ export namespace coreServices { * @public */ export const scheduler = createServiceRef< - import('@backstage/backend-plugin-api/scheduler').SchedulerService + import('./SchedulerService').SchedulerService >({ id: 'core.scheduler' }); /** diff --git a/packages/backend-plugin-api/src/services/definitions/index.ts b/packages/backend-plugin-api/src/services/definitions/index.ts index bd436cf899..5739add90c 100644 --- a/packages/backend-plugin-api/src/services/definitions/index.ts +++ b/packages/backend-plugin-api/src/services/definitions/index.ts @@ -52,6 +52,16 @@ export type { PluginMetadataService } from './PluginMetadataService'; export type { RootHttpRouterService } from './RootHttpRouterService'; export type { RootLifecycleService } from './RootLifecycleService'; export type { RootLoggerService } from './RootLoggerService'; +export { readSchedulerServiceTaskScheduleDefinitionFromConfig } from './SchedulerService'; +export type { + SchedulerService, + SchedulerServiceTaskDescriptor, + SchedulerServiceTaskFunction, + SchedulerServiceTaskInvocationDefinition, + SchedulerServiceTaskRunner, + SchedulerServiceTaskScheduleDefinition, + SchedulerServiceTaskScheduleDefinitionConfig, +} from './SchedulerService'; export type { TokenManagerService } from './TokenManagerService'; export type { ReadTreeOptions, diff --git a/packages/backend-tasks/src/tasks/TaskScheduler.ts b/packages/backend-tasks/src/tasks/TaskScheduler.ts index ac9039adcd..daff280020 100644 --- a/packages/backend-tasks/src/tasks/TaskScheduler.ts +++ b/packages/backend-tasks/src/tasks/TaskScheduler.ts @@ -33,7 +33,7 @@ import { LoggerService } from '@backstage/backend-plugin-api'; * Deals with the scheduling of distributed tasks. * * @public - * @deprecated Please migrate to the new backend system, and depend on `coreServices.scheduler` from `@backstage/backend-plugin-api` instead. The new default implementation of this service lives in `@backstage/backend-defaults/scheduler`. + * @deprecated Please migrate to the new backend system, and depend on `coreServices.scheduler` from `@backstage/backend-plugin-api` instead. */ export class TaskScheduler { static fromConfig( diff --git a/packages/backend-tasks/src/tasks/readTaskScheduleDefinitionFromConfig.ts b/packages/backend-tasks/src/tasks/readTaskScheduleDefinitionFromConfig.ts index 2e761f2f94..5a173f246b 100644 --- a/packages/backend-tasks/src/tasks/readTaskScheduleDefinitionFromConfig.ts +++ b/packages/backend-tasks/src/tasks/readTaskScheduleDefinitionFromConfig.ts @@ -51,7 +51,7 @@ function readCronOrDuration( * * @param config - config for a TaskScheduleDefinition. * @public - * @deprecated Please import from `@backstage/backend-plugin-api/scheduler` instead + * @deprecated Please import `readSchedulerServiceTaskScheduleDefinitionFromConfig` from `@backstage/backend-plugin-api` instead */ export function readTaskScheduleDefinitionFromConfig( config: Config, diff --git a/packages/backend-tasks/src/tasks/types.ts b/packages/backend-tasks/src/tasks/types.ts index 3d49e19cc1..955909e040 100644 --- a/packages/backend-tasks/src/tasks/types.ts +++ b/packages/backend-tasks/src/tasks/types.ts @@ -26,7 +26,7 @@ import { z } from 'zod'; * processing should abort and return as quickly as possible. * * @public - * @deprecated Please import from `@backstage/backend-plugin-api/scheduler` instead + * @deprecated Please import `SchedulerServiceTaskFunction` from `@backstage/backend-plugin-api` instead */ export type TaskFunction = | ((abortSignal: AbortSignal) => void | Promise) @@ -36,7 +36,7 @@ export type TaskFunction = * A semi-opaque type to describe an actively scheduled task. * * @public - * @deprecated Please import from `@backstage/backend-plugin-api/scheduler` instead + * @deprecated Please import `SchedulerServiceTaskDescriptor` from `@backstage/backend-plugin-api` instead */ export type TaskDescriptor = { /** @@ -59,7 +59,7 @@ export type TaskDescriptor = { * Options that control the scheduling of a task. * * @public - * @deprecated Please import from `@backstage/backend-plugin-api/scheduler` instead + * @deprecated Please import `SchedulerServiceTaskScheduleDefinition` from `@backstage/backend-plugin-api` instead */ export interface TaskScheduleDefinition { /** @@ -157,7 +157,7 @@ export interface TaskScheduleDefinition { * that control the scheduling of a task. * * @public - * @deprecated Please import from `@backstage/backend-plugin-api/scheduler` instead + * @deprecated Please import `SchedulerServiceTaskDefinitionConfig` from `@backstage/backend-plugin-api` instead */ export interface TaskScheduleDefinitionConfig { /** @@ -254,7 +254,7 @@ export interface TaskScheduleDefinitionConfig { * Options that apply to the invocation of a given task. * * @public - * @deprecated Please import from `@backstage/backend-plugin-api/scheduler` instead + * @deprecated Please import `SchedulerServiceTaskInvocationDefinition` from `@backstage/backend-plugin-api` instead */ export interface TaskInvocationDefinition { /** @@ -278,7 +278,7 @@ export interface TaskInvocationDefinition { * A previously prepared task schedule, ready to be invoked. * * @public - * @deprecated Please import from `@backstage/backend-plugin-api/scheduler` instead + * @deprecated Please import `SchedulerServiceTaskRunner` from `@backstage/backend-plugin-api` instead */ export interface TaskRunner { /** @@ -293,7 +293,7 @@ export interface TaskRunner { * Deals with the scheduling of distributed tasks, for a given plugin. * * @public - * @deprecated Please use `SchedulerService` from `@backstage/backend-plugin-api/scheduler` instead + * @deprecated Please use `SchedulerService` from `@backstage/backend-plugin-api` instead (most likely via `coreServices.scheduler`) */ export interface PluginTaskScheduler { /** diff --git a/packages/backend-test-utils/api-report.md b/packages/backend-test-utils/api-report.md index 6e9cdc9459..4f6eb47243 100644 --- a/packages/backend-test-utils/api-report.md +++ b/packages/backend-test-utils/api-report.md @@ -34,7 +34,7 @@ import { RootHttpRouterFactoryOptions } from '@backstage/backend-app-api'; import { RootHttpRouterService } from '@backstage/backend-plugin-api'; import { RootLifecycleService } from '@backstage/backend-plugin-api'; import { RootLoggerService } from '@backstage/backend-plugin-api'; -import { SchedulerService } from '@backstage/backend-plugin-api/scheduler'; +import { SchedulerService } from '@backstage/backend-plugin-api'; import { ServiceFactory } from '@backstage/backend-plugin-api'; import { ServiceRef } from '@backstage/backend-plugin-api'; import { TokenManagerService } from '@backstage/backend-plugin-api'; diff --git a/yarn.lock b/yarn.lock index 3bdc15bbb7..23620ac274 100644 --- a/yarn.lock +++ b/yarn.lock @@ -3433,7 +3433,17 @@ __metadata: "@backstage/backend-plugin-api": "workspace:^" "@backstage/backend-test-utils": "workspace:^" "@backstage/cli": "workspace:^" + "@backstage/errors": "workspace:^" "@backstage/plugin-events-node": "workspace:^" + "@backstage/types": "workspace:^" + "@opentelemetry/api": ^1.3.0 + cron: ^3.0.0 + knex: ^3.0.0 + lodash: ^4.17.21 + luxon: ^3.0.0 + uuid: ^9.0.0 + wait-for-expect: ^3.0.2 + zod: ^3.22.4 languageName: unknown linkType: soft