diff --git a/.changeset/happy-rice-tickle.md b/.changeset/happy-rice-tickle.md new file mode 100644 index 0000000000..475f03501c --- /dev/null +++ b/.changeset/happy-rice-tickle.md @@ -0,0 +1,5 @@ +--- +'@backstage/backend-common': patch +--- + +Add support for distributed mutexes and scheduled tasks, through the `TaskManager` class. This class can be particularly useful for coordinating things across many deployed instances of a given backend plugin. An example of this is catalog entity providers - with this facility you can register tasks similar to a cron job, and make sure that only one host at a time tries to execute the job, and that the timing (call frequency, timeouts etc) are retained as a global concern, letting you scale your workload safely without affecting the task behavior. diff --git a/.github/styles/vocab.txt b/.github/styles/vocab.txt index 7e478953bc..c14047cd4d 100644 --- a/.github/styles/vocab.txt +++ b/.github/styles/vocab.txt @@ -52,6 +52,7 @@ configmaps configs const cookiecutter +cron css Datadog dataflow @@ -162,6 +163,8 @@ Monorepo monorepos msgraph msw +mutex +mutexes mysql namespace namespaced diff --git a/packages/backend-common/api-report.md b/packages/backend-common/api-report.md index 480b6419fa..688c74159e 100644 --- a/packages/backend-common/api-report.md +++ b/packages/backend-common/api-report.md @@ -384,6 +384,11 @@ export function loadBackendConfig(options: { argv: string[]; }): Promise; +// @public +export interface LockOptions { + timeout: Duration; +} + // @public export function notFoundHandler(): RequestHandler; @@ -405,30 +410,22 @@ export type PluginEndpointDiscovery = { // @public export interface PluginTaskManager { - // (undocumented) acquireLock( id: string, - options: { - timeout: Duration; - }, + options: LockOptions, ): Promise< | { acquired: false; } | { acquired: true; - release: () => void | Promise; + release(): Promise; } >; - // (undocumented) scheduleTask( id: string, - options: { - timeout: Duration; - frequency: Duration; - initialDelay?: Duration; - }, - fn: () => Promise, + options: TaskOptions, + fn: () => void | Promise, ): Promise<{ unschedule: () => Promise; }>; @@ -614,7 +611,6 @@ export interface StatusCheckHandlerOptions { // @public export class TaskManager { constructor(databaseManager: DatabaseManager, logger: Logger_2); - // (undocumented) forPlugin(pluginId: string): PluginTaskManager; // (undocumented) static fromConfig( @@ -626,6 +622,13 @@ export class TaskManager { ): TaskManager; } +// @public +export interface TaskOptions { + frequency?: Duration; + initialDelay?: Duration; + timeout?: Duration; +} + // @public export type UrlReader = { read(url: string): Promise; diff --git a/packages/backend-common/package.json b/packages/backend-common/package.json index e0d4be1242..555c2004ee 100644 --- a/packages/backend-common/package.json +++ b/packages/backend-common/package.json @@ -90,6 +90,7 @@ "@types/concat-stream": "^1.6.0", "@types/fs-extra": "^9.0.3", "@types/http-errors": "^1.6.3", + "@types/luxon": "^2.0.4", "@types/minimist": "^1.2.0", "@types/mock-fs": "^4.13.0", "@types/morgan": "^1.9.0", diff --git a/packages/backend-common/src/tasks/PluginTaskManagerImpl.ts b/packages/backend-common/src/tasks/PluginTaskManagerImpl.ts index 2d1f192fe6..540c7d501a 100644 --- a/packages/backend-common/src/tasks/PluginTaskManagerImpl.ts +++ b/packages/backend-common/src/tasks/PluginTaskManagerImpl.ts @@ -15,13 +15,12 @@ */ import { Knex } from 'knex'; -import { Duration } from 'luxon'; import { v4 as uuid } from 'uuid'; import { Logger } from 'winston'; import { isDatabaseConflictError } from '../database'; import { DbMutexesRow, DB_MUTEXES_TABLE } from '../database/tables'; import { TaskWorker } from './TaskWorker'; -import { PluginTaskManager } from './types'; +import { LockOptions, PluginTaskManager, TaskOptions } from './types'; import { nowPlus, validateId } from './util'; /** @@ -35,12 +34,9 @@ export class PluginTaskManagerImpl implements PluginTaskManager { async acquireLock( id: string, - options: { - timeout: Duration; - }, + options: LockOptions, ): Promise< - | { acquired: false } - | { acquired: true; release: () => void | Promise } + { acquired: false } | { acquired: true; release(): Promise } > { validateId(id); @@ -90,11 +86,7 @@ export class PluginTaskManagerImpl implements PluginTaskManager { async scheduleTask( id: string, - options: { - timeout?: Duration; - frequency?: Duration; - initialDelay?: Duration; - }, + options: TaskOptions, fn: () => void | Promise, ): Promise<{ unschedule: () => Promise }> { validateId(id); diff --git a/packages/backend-common/src/tasks/TaskManager.ts b/packages/backend-common/src/tasks/TaskManager.ts index ea49c2383e..db8228561f 100644 --- a/packages/backend-common/src/tasks/TaskManager.ts +++ b/packages/backend-common/src/tasks/TaskManager.ts @@ -51,6 +51,12 @@ export class TaskManager { private readonly logger: Logger, ) {} + /** + * Instantiates a task manager instance for the given plugin. + * + * @param pluginId - The unique ID of the plugin, for example "catalog" + * @returns A {@link PluginTaskManager} instance + */ forPlugin(pluginId: string): PluginTaskManager { const databaseFactory = memoize(async () => { const knex = await this.databaseManager.forPlugin(pluginId).getClient(); diff --git a/packages/backend-common/src/tasks/index.ts b/packages/backend-common/src/tasks/index.ts index 60329788b3..9f86f8fc64 100644 --- a/packages/backend-common/src/tasks/index.ts +++ b/packages/backend-common/src/tasks/index.ts @@ -14,5 +14,5 @@ * limitations under the License. */ -export type { PluginTaskManager } from './types'; export { TaskManager } from './TaskManager'; +export type { LockOptions, PluginTaskManager, TaskOptions } from './types'; diff --git a/packages/backend-common/src/tasks/types.ts b/packages/backend-common/src/tasks/types.ts index 4b061f9069..a8d356bed9 100644 --- a/packages/backend-common/src/tasks/types.ts +++ b/packages/backend-common/src/tasks/types.ts @@ -17,6 +17,64 @@ import { Duration } from 'luxon'; import { z } from 'zod'; +/** + * Options that apply to the acquiral of a given lock. + * + * @public + */ +export interface LockOptions { + /** + * The maximum amount of time that the lock can be held, before it's + * considered timed out and gets auto-released by the framework. + */ + timeout: Duration; +} + +/** + * Options that apply to the invocation of a given task. + * + * @public + */ +export interface TaskOptions { + /** + * The maximum amount of time that a single task invocation can take, before + * it's considered timed out and gets "released" such that a new invocation + * is permitted to take place (possibly, then, on a different worker). + * + * If no value is given for this field then there is no timeout. This is + * potentially dangerous. + */ + timeout?: Duration; + + /** + * The amount of time that should pass between task invocation starts. + * Essentially, this equals roughly how often you want the task to run. + * + * This is a best effort value; under some circumstances there can be + * deviations. For example, if the task runtime is longer than the frequency + * and the timeout has not been given or not been exceeded yet, the next + * invocation of this task will be delayed until after the previous one + * finishes. + * + * The system does its best to avoid overlapping invocations. + * + * If no value is given for this field then the task will only be invoked + * once (on any worker) and then unscheduled automatically. + */ + frequency?: Duration; + + /** + * The amount of time that should pass before the first invocation happens. + * + * This can be useful in cold start scenarios to stagger or delay some heavy + * compute jobs. + * + * If no value is given for this field then the first invocation will happen + * as soon as possible. + */ + initialDelay?: Duration; +} + /** * Deals with management and locking related to distributed tasks, for a given * plugin. @@ -24,24 +82,45 @@ import { z } from 'zod'; * @public */ export interface PluginTaskManager { + /** + * Attempts to acquire an exclusive lock. + * + * A lock can only be held by one party at a time. Any subsequent attempts to + * acquire the lock will fail, unless the timeout period has been exceeded or + * the lock was released by the previous holder. + * + * @param id - A unique ID (within the scope of the plugin) for a lock + * @param options - Options for the lock + * @returns The result of the lock attempt. If it was successfully acquired, + * you should remember to call its `release` method as soon as you + * are done with the lock. + */ acquireLock( id: string, - options: { - timeout: Duration; - }, + options: LockOptions, ): Promise< - | { acquired: false } - | { acquired: true; release: () => void | Promise } + { acquired: false } | { acquired: true; release(): Promise } >; + /** + * Schedules a task function for coordinated exclusive invocation across + * workers. + * + * If the task was already scheduled since before by us or by another party, + * its options are just overwritten with the given options, and things + * continue from there. + * + * @param id - A unique ID (within the scope of the plugin) for the task + * @param options - Options for the task + * @param fn - The actual task function to be invoked + * @returns An `unschedule` function that can be used to stop the task + * invocations later on. This removes the task entirely from storage + * and stops its invocations across all workers. + */ scheduleTask( id: string, - options: { - timeout: Duration; - frequency: Duration; - initialDelay?: Duration; - }, - fn: () => Promise, + options: TaskOptions, + fn: () => void | Promise, ): Promise<{ unschedule: () => Promise }>; }