move to a separate package instead and address comments
Signed-off-by: Fredrik Adelöw <freben@gmail.com>
This commit is contained in:
Fredrik Adelöw
@@ -0,0 +1,3 @@
|
||||
module.exports = {
|
||||
extends: [require.resolve('@backstage/cli/config/eslint.backend')],
|
||||
};
|
||||
@@ -0,0 +1,36 @@
|
||||
# @backstage/backend-tasks
|
||||
|
||||
Common distributed task management / locking library for Backstage backends.
|
||||
|
||||
## Usage
|
||||
|
||||
Add the library to your backend package:
|
||||
|
||||
```sh
|
||||
# From your Backstage root directory
|
||||
cd packages/backend
|
||||
yarn add @backstage/backend-tasks
|
||||
```
|
||||
|
||||
then make use of its facilities as necessary:
|
||||
|
||||
```typescript
|
||||
import { TaskManager } from '@backstage/backend-tasks';
|
||||
|
||||
const manager = TaskManager.fromConfig(rootConfig).forPlugin('my-plugin');
|
||||
|
||||
const { unschedule } = await manager.scheduleTask(
|
||||
'refresh-things',
|
||||
{
|
||||
frequency: Duration.fromObject({ minutes: 10 }),
|
||||
},
|
||||
async () => {
|
||||
await entityProvider.run();
|
||||
},
|
||||
);
|
||||
```
|
||||
|
||||
## Documentation
|
||||
|
||||
- [Backstage Readme](https://github.com/backstage/backstage/blob/master/README.md)
|
||||
- [Backstage Documentation](https://github.com/backstage/backstage/blob/master/docs/README.md)
|
||||
@@ -0,0 +1,59 @@
|
||||
## API Report File for "@backstage/backend-tasks"
|
||||
|
||||
> Do not edit this file. It is a report generated by [API Extractor](https://api-extractor.com/).
|
||||
|
||||
```ts
|
||||
import { Config } from '@backstage/config';
|
||||
import { DatabaseManager } from '@backstage/backend-common';
|
||||
import { Duration } from 'luxon';
|
||||
import { Logger as Logger_2 } from 'winston';
|
||||
|
||||
// @public
|
||||
export interface LockOptions {
|
||||
timeout: Duration;
|
||||
}
|
||||
|
||||
// @public
|
||||
export interface PluginTaskManager {
|
||||
acquireLock(
|
||||
id: string,
|
||||
options: LockOptions,
|
||||
): Promise<
|
||||
| {
|
||||
acquired: false;
|
||||
}
|
||||
| {
|
||||
acquired: true;
|
||||
release(): Promise<void>;
|
||||
}
|
||||
>;
|
||||
scheduleTask(
|
||||
id: string,
|
||||
options: TaskOptions,
|
||||
fn: () => void | Promise<void>,
|
||||
): Promise<{
|
||||
unschedule: () => Promise<void>;
|
||||
}>;
|
||||
}
|
||||
|
||||
// @public
|
||||
export class TaskManager {
|
||||
constructor(databaseManager: DatabaseManager, logger: Logger_2);
|
||||
forPlugin(pluginId: string): PluginTaskManager;
|
||||
// (undocumented)
|
||||
static fromConfig(
|
||||
config: Config,
|
||||
options?: {
|
||||
databaseManager?: DatabaseManager;
|
||||
logger?: Logger_2;
|
||||
},
|
||||
): TaskManager;
|
||||
}
|
||||
|
||||
// @public
|
||||
export interface TaskOptions {
|
||||
frequency?: Duration;
|
||||
initialDelay?: Duration;
|
||||
timeout?: Duration;
|
||||
}
|
||||
```
|
||||
@@ -0,0 +1,26 @@
|
||||
/*
|
||||
* 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.
|
||||
*/
|
||||
|
||||
// This file makes it possible to run "yarn knex migrate:make some_file_name"
|
||||
// to assist in making new migrations
|
||||
module.exports = {
|
||||
client: 'sqlite3',
|
||||
connection: ':memory:',
|
||||
useNullAsDefault: true,
|
||||
migrations: {
|
||||
directory: './migrations',
|
||||
},
|
||||
};
|
||||
@@ -0,0 +1,99 @@
|
||||
/*
|
||||
* 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) {
|
||||
//
|
||||
// mutexes
|
||||
//
|
||||
await knex.schema.createTable('backstage_backend_tasks__mutexes', table => {
|
||||
table.comment('Locks used for mutual exclusion among multiple workers');
|
||||
table
|
||||
.text('id')
|
||||
.primary()
|
||||
.notNullable()
|
||||
.comment('The unique ID of this particular mutex');
|
||||
table
|
||||
.text('current_lock_ticket')
|
||||
.notNullable()
|
||||
.comment('A unique ticket for the current mutex lock');
|
||||
table
|
||||
.dateTime('current_lock_acquired_at')
|
||||
.nullable()
|
||||
.comment('The time when the mutex was locked');
|
||||
table
|
||||
.dateTime('current_lock_expires_at')
|
||||
.nullable()
|
||||
.comment('The time when a locked mutex will time out and auto-release');
|
||||
table.index(['id'], 'backstage_backend_tasks__mutexes__id_idx');
|
||||
});
|
||||
//
|
||||
// tasks
|
||||
//
|
||||
await knex.schema.createTable('backstage_backend_tasks__tasks', table => {
|
||||
table.comment('Tasks used for scheduling work on multiple workers');
|
||||
table
|
||||
.text('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')
|
||||
.nullable()
|
||||
.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');
|
||||
table.index(['id'], 'backstage_backend_tasks__tasks__id_idx');
|
||||
});
|
||||
};
|
||||
|
||||
/**
|
||||
* @param {import('knex').Knex} knex
|
||||
*/
|
||||
exports.down = async function down(knex) {
|
||||
//
|
||||
// tasks
|
||||
//
|
||||
await knex.schema.alterTable('backstage_backend_tasks__tasks', table => {
|
||||
table.dropIndex([], 'backstage_backend_tasks__tasks__id_idx');
|
||||
});
|
||||
await knex.schema.dropTable('backstage_backend_tasks__tasks');
|
||||
//
|
||||
// locks
|
||||
//
|
||||
await knex.schema.alterTable('backstage_backend_tasks__task_locks', table => {
|
||||
table.dropIndex([], 'backstage_backend_tasks__task_locks__id_idx');
|
||||
});
|
||||
await knex.schema.dropTable('backstage_backend_tasks__task_locks');
|
||||
};
|
||||
@@ -0,0 +1,54 @@
|
||||
{
|
||||
"name": "@backstage/backend-tasks",
|
||||
"description": "Common distributed task management / locking library for Backstage backends",
|
||||
"version": "0.1.0",
|
||||
"main": "src/index.ts",
|
||||
"types": "src/index.ts",
|
||||
"private": false,
|
||||
"publishConfig": {
|
||||
"access": "public",
|
||||
"main": "dist/index.cjs.js",
|
||||
"types": "dist/index.d.ts"
|
||||
},
|
||||
"homepage": "https://backstage.io",
|
||||
"repository": {
|
||||
"type": "git",
|
||||
"url": "https://github.com/backstage/backstage",
|
||||
"directory": "packages/backend-tasks"
|
||||
},
|
||||
"keywords": [
|
||||
"backstage"
|
||||
],
|
||||
"license": "Apache-2.0",
|
||||
"scripts": {
|
||||
"build": "backstage-cli build --outputs cjs,types",
|
||||
"lint": "backstage-cli lint",
|
||||
"test": "backstage-cli test",
|
||||
"prepack": "backstage-cli prepack",
|
||||
"postpack": "backstage-cli postpack",
|
||||
"clean": "backstage-cli clean"
|
||||
},
|
||||
"dependencies": {
|
||||
"@backstage/backend-common": "^0.9.7",
|
||||
"@backstage/config": "^0.1.10",
|
||||
"@backstage/errors": "^0.1.3",
|
||||
"@backstage/types": "^0.1.1",
|
||||
"@types/luxon": "^2.0.4",
|
||||
"knex": "^0.95.1",
|
||||
"lodash": "^4.17.21",
|
||||
"luxon": "^2.0.2",
|
||||
"uuid": "^8.0.0",
|
||||
"winston": "^3.2.1",
|
||||
"zod": "^3.9.5"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@backstage/backend-test-utils": "^0.1.7",
|
||||
"@backstage/cli": "^0.8.0",
|
||||
"jest": "^26.0.1",
|
||||
"wait-for-expect": "^3.0.2"
|
||||
},
|
||||
"files": [
|
||||
"dist",
|
||||
"migrations/**/*.{js,d.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';
|
||||
|
||||
const migrationsDir = resolvePackagePath(
|
||||
'@backstage/backend-tasks',
|
||||
'migrations',
|
||||
);
|
||||
|
||||
export async function migrateBackendTasks(knex: Knex): Promise<void> {
|
||||
await knex.migrate.latest({
|
||||
directory: migrationsDir,
|
||||
tableName: DB_MIGRATIONS_TABLE,
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
/*
|
||||
* 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_MUTEXES_TABLE = 'backstage_backend_tasks__mutexes';
|
||||
export const DB_TASKS_TABLE = 'backstage_backend_tasks__tasks';
|
||||
|
||||
export type DbMutexesRow = {
|
||||
id: string;
|
||||
current_lock_ticket: string;
|
||||
current_lock_acquired_at?: Date | string;
|
||||
current_lock_expires_at?: Date | string;
|
||||
};
|
||||
|
||||
export type DbTasksRow = {
|
||||
id: string;
|
||||
settings_json: string;
|
||||
next_run_start_at?: Date | string;
|
||||
current_run_ticket?: string;
|
||||
current_run_started_at?: Date | string;
|
||||
current_run_expires_at?: Date | string;
|
||||
};
|
||||
@@ -0,0 +1,23 @@
|
||||
/*
|
||||
* 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.
|
||||
*/
|
||||
|
||||
/**
|
||||
* Common distributed task management / locking library for Backstage backends
|
||||
*
|
||||
* @packageDocumentation
|
||||
*/
|
||||
|
||||
export * from './tasks';
|
||||
@@ -0,0 +1,17 @@
|
||||
/*
|
||||
* 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.
|
||||
*/
|
||||
|
||||
export {};
|
||||
@@ -0,0 +1,48 @@
|
||||
/*
|
||||
* 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 class CancelToken {
|
||||
// @ts-ignore: is actually assigned by the Promise constructor
|
||||
#cancel: () => void;
|
||||
#isCancelled: boolean;
|
||||
#cancelPromise: Promise<void>;
|
||||
|
||||
static create(): CancelToken {
|
||||
return new CancelToken();
|
||||
}
|
||||
|
||||
private constructor() {
|
||||
this.#isCancelled = false;
|
||||
this.#cancelPromise = new Promise(resolve => {
|
||||
this.#cancel = () => {
|
||||
this.#isCancelled = true;
|
||||
resolve();
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
cancel(): void {
|
||||
this.#cancel();
|
||||
}
|
||||
|
||||
get isCancelled(): boolean {
|
||||
return this.#isCancelled;
|
||||
}
|
||||
|
||||
get promise(): Promise<void> {
|
||||
return this.#cancelPromise;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,138 @@
|
||||
/*
|
||||
* 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 { Duration } from 'luxon';
|
||||
import waitForExpect from 'wait-for-expect';
|
||||
import { migrateBackendTasks } from '../database/migrateBackendTasks';
|
||||
import { PluginTaskManagerImpl } from './PluginTaskManagerImpl';
|
||||
|
||||
describe('PluginTaskManagerImpl', () => {
|
||||
const databases = TestDatabases.create({
|
||||
ids: ['POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'],
|
||||
});
|
||||
|
||||
async function init(databaseId: TestDatabaseId) {
|
||||
const knex = await databases.init(databaseId);
|
||||
await migrateBackendTasks(knex);
|
||||
const manager = new PluginTaskManagerImpl(
|
||||
async () => knex,
|
||||
getVoidLogger(),
|
||||
);
|
||||
return { knex, manager };
|
||||
}
|
||||
|
||||
describe('acquireLock', () => {
|
||||
it.each(databases.eachSupportedId())(
|
||||
'can run the happy path, %p',
|
||||
async databaseId => {
|
||||
const { manager } = await init(databaseId);
|
||||
|
||||
const lock1 = await manager.acquireLock('lock1', {
|
||||
timeout: Duration.fromMillis(5000),
|
||||
});
|
||||
const lock2 = await manager.acquireLock('lock2', {
|
||||
timeout: Duration.fromMillis(5000),
|
||||
});
|
||||
|
||||
expect(lock1.acquired).toBe(true);
|
||||
expect(lock2.acquired).toBe(true);
|
||||
|
||||
await expect(
|
||||
manager.acquireLock('lock1', {
|
||||
timeout: Duration.fromMillis(5000),
|
||||
}),
|
||||
).resolves.toEqual({ acquired: false });
|
||||
|
||||
await (lock1 as any).release();
|
||||
await (lock2 as any).release();
|
||||
|
||||
const lock1Again = await manager.acquireLock('lock1', {
|
||||
timeout: Duration.fromMillis(5000),
|
||||
});
|
||||
expect(lock1Again.acquired).toBe(true);
|
||||
await (lock1Again as any).release();
|
||||
},
|
||||
);
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'rejects double lock attempts, %p',
|
||||
async databaseId => {
|
||||
const { manager } = await init(databaseId);
|
||||
|
||||
const lock1 = await manager.acquireLock('lock1', {
|
||||
timeout: Duration.fromMillis(5000),
|
||||
});
|
||||
const lock2 = await manager.acquireLock('lock1', {
|
||||
timeout: Duration.fromMillis(5000),
|
||||
});
|
||||
|
||||
expect(lock1.acquired).toBe(true);
|
||||
expect(lock2.acquired).toBe(false);
|
||||
|
||||
await (lock1 as any).release();
|
||||
|
||||
const lock1Again = await manager.acquireLock('lock1', {
|
||||
timeout: Duration.fromMillis(5000),
|
||||
});
|
||||
expect(lock1Again.acquired).toBe(true);
|
||||
await (lock1Again as any).release();
|
||||
},
|
||||
);
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'times out locks, %p',
|
||||
async databaseId => {
|
||||
const { manager } = await init(databaseId);
|
||||
|
||||
const lock1 = await manager.acquireLock('lock1', {
|
||||
timeout: Duration.fromMillis(200),
|
||||
});
|
||||
|
||||
expect(lock1.acquired).toBe(true);
|
||||
|
||||
await new Promise(resolve => setTimeout(resolve, 1000));
|
||||
|
||||
const lock2 = await manager.acquireLock('lock1', {
|
||||
timeout: Duration.fromMillis(5000),
|
||||
});
|
||||
expect(lock2.acquired).toBe(true);
|
||||
await (lock2 as any).release();
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
// This is just to test the wrapper code; most of the actual tests are in
|
||||
// TaskWorker.test.ts
|
||||
describe('scheduleTask', () => {
|
||||
it.each(databases.eachSupportedId())(
|
||||
'can run the happy path, %p',
|
||||
async databaseId => {
|
||||
const { manager } = await init(databaseId);
|
||||
|
||||
const fn = jest.fn();
|
||||
const { unschedule } = await manager.scheduleTask('task1', {}, fn);
|
||||
|
||||
await waitForExpect(() => {
|
||||
expect(fn).toBeCalled();
|
||||
});
|
||||
|
||||
await unschedule();
|
||||
},
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,109 @@
|
||||
/*
|
||||
* 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 { isDatabaseConflictError } from '@backstage/backend-common';
|
||||
import { Knex } from 'knex';
|
||||
import { v4 as uuid } from 'uuid';
|
||||
import { Logger } from 'winston';
|
||||
import { DbMutexesRow, DB_MUTEXES_TABLE } from '../database/tables';
|
||||
import { TaskWorker } from './TaskWorker';
|
||||
import { LockOptions, PluginTaskManager, TaskOptions } from './types';
|
||||
import { nowPlus, validateId } from './util';
|
||||
|
||||
/**
|
||||
* Implements the actual task management.
|
||||
*/
|
||||
export class PluginTaskManagerImpl implements PluginTaskManager {
|
||||
constructor(
|
||||
private readonly databaseFactory: () => Promise<Knex>,
|
||||
private readonly logger: Logger,
|
||||
) {}
|
||||
|
||||
async acquireLock(
|
||||
id: string,
|
||||
options: LockOptions,
|
||||
): Promise<
|
||||
{ acquired: false } | { acquired: true; release(): Promise<void> }
|
||||
> {
|
||||
validateId(id);
|
||||
|
||||
const knex = await this.databaseFactory();
|
||||
const ticket = uuid();
|
||||
|
||||
async function release() {
|
||||
try {
|
||||
await knex<DbMutexesRow>(DB_MUTEXES_TABLE)
|
||||
.where('id', '=', id)
|
||||
.where('current_lock_ticket', '=', ticket)
|
||||
.delete();
|
||||
} catch {
|
||||
// fail silently
|
||||
}
|
||||
}
|
||||
|
||||
const record: Knex.DbRecord<DbMutexesRow> = {
|
||||
current_lock_ticket: ticket,
|
||||
current_lock_acquired_at: knex.fn.now(),
|
||||
current_lock_expires_at: options.timeout
|
||||
? nowPlus(options.timeout, knex)
|
||||
: knex.raw('null'),
|
||||
};
|
||||
|
||||
// First try to overwrite an existing lock, that has timed out
|
||||
const stolen = await knex<DbMutexesRow>(DB_MUTEXES_TABLE)
|
||||
.where('id', '=', id)
|
||||
.where('current_lock_expires_at', '<', knex.fn.now())
|
||||
.update(record);
|
||||
|
||||
if (stolen) {
|
||||
return { acquired: true, release };
|
||||
}
|
||||
|
||||
try {
|
||||
await knex<DbMutexesRow>(DB_MUTEXES_TABLE).insert({ id, ...record });
|
||||
return { acquired: true, release };
|
||||
} catch (e) {
|
||||
if (!isDatabaseConflictError(e)) {
|
||||
this.logger.warn(`Failed to acquire lock, ${e}`);
|
||||
}
|
||||
return { acquired: false };
|
||||
}
|
||||
}
|
||||
|
||||
async scheduleTask(
|
||||
id: string,
|
||||
options: TaskOptions,
|
||||
fn: () => void | Promise<void>,
|
||||
): Promise<{ unschedule: () => Promise<void> }> {
|
||||
validateId(id);
|
||||
|
||||
const knex = await this.databaseFactory();
|
||||
|
||||
const task = new TaskWorker(id, fn, knex, this.logger);
|
||||
await task.start({
|
||||
version: 1,
|
||||
initialDelayDuration: options.initialDelay?.toISO(),
|
||||
recurringAtMostEveryDuration: options.frequency?.toISO(),
|
||||
timeoutAfterDuration: options.timeout?.toISO(),
|
||||
});
|
||||
|
||||
return {
|
||||
async unschedule() {
|
||||
await task.stop();
|
||||
},
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,114 @@
|
||||
/*
|
||||
* 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 { Knex } from 'knex';
|
||||
import { Duration } from 'luxon';
|
||||
import { Logger } from 'winston';
|
||||
import {
|
||||
DbMutexesRow,
|
||||
DbTasksRow,
|
||||
DB_MUTEXES_TABLE,
|
||||
DB_TASKS_TABLE,
|
||||
} from '../database/tables';
|
||||
import { CancelToken } from './CancelToken';
|
||||
|
||||
/**
|
||||
* Makes sure to auto-expire and clean up things that time out or for other
|
||||
* reasons should not be left lingering.
|
||||
*/
|
||||
export class PluginTaskManagerJanitor {
|
||||
private readonly knex: Knex;
|
||||
private readonly waitBetweenRuns: Duration;
|
||||
private readonly logger: Logger;
|
||||
private readonly cancelToken: CancelToken;
|
||||
|
||||
constructor(options: {
|
||||
knex: Knex;
|
||||
waitBetweenRuns: Duration;
|
||||
logger: Logger;
|
||||
}) {
|
||||
this.knex = options.knex;
|
||||
this.waitBetweenRuns = options.waitBetweenRuns;
|
||||
this.logger = options.logger;
|
||||
this.cancelToken = CancelToken.create();
|
||||
}
|
||||
|
||||
async start() {
|
||||
while (!this.cancelToken.isCancelled) {
|
||||
try {
|
||||
await this.runOnce();
|
||||
} catch (e) {
|
||||
this.logger.warn(`Error while performing janitorial tasks, ${e}`);
|
||||
}
|
||||
|
||||
await this.sleep(this.waitBetweenRuns);
|
||||
}
|
||||
}
|
||||
|
||||
async stop() {
|
||||
this.cancelToken.cancel();
|
||||
}
|
||||
|
||||
private async runOnce() {
|
||||
// SQLite currently (Oct 1 2021) returns a number for returning()
|
||||
// statements, effectively ignoring them and instead returning the outcome
|
||||
// of the delete() - and knex also emits a warning about that fact, which
|
||||
// is why we avoid that entirely for the sqlite3 driver.
|
||||
// https://github.com/knex/knex/issues/4370
|
||||
// https://github.com/mapbox/node-sqlite3/issues/1453
|
||||
|
||||
const mutexesQuery = this.knex<DbMutexesRow>(DB_MUTEXES_TABLE)
|
||||
.where('current_lock_expires_at', '<', this.knex.fn.now())
|
||||
.delete();
|
||||
|
||||
if (this.knex.client.config.client === 'sqlite3') {
|
||||
const mutexes = await mutexesQuery;
|
||||
if (mutexes > 0) {
|
||||
this.logger.warn(`${mutexes} mutex locks timed out and were lost`);
|
||||
}
|
||||
} else {
|
||||
const mutexes = await mutexesQuery.returning(['id']);
|
||||
for (const { id } of mutexes) {
|
||||
this.logger.warn(`Mutex lock timed out and was lost: ${id}`);
|
||||
}
|
||||
}
|
||||
|
||||
const tasksQuery = this.knex<DbTasksRow>(DB_TASKS_TABLE)
|
||||
.where('current_run_expires_at', '<', this.knex.fn.now())
|
||||
.delete();
|
||||
|
||||
if (this.knex.client.config.client === 'sqlite3') {
|
||||
const tasks = await tasksQuery;
|
||||
this.logger.warn(`${tasks} tasks timed out and were lost`);
|
||||
} else {
|
||||
const tasks = await tasksQuery.returning(['id']);
|
||||
for (const { id } of tasks) {
|
||||
this.logger.warn(`Task timed out and was lost: ${id}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Sleeps for the given duration, but aborts sooner if the cancel token
|
||||
* triggers.
|
||||
*/
|
||||
private async sleep(duration: Duration) {
|
||||
await Promise.race([
|
||||
new Promise(resolve => setTimeout(resolve, duration.as('milliseconds'))),
|
||||
this.cancelToken.promise,
|
||||
]);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,52 @@
|
||||
/*
|
||||
* 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 { TaskManager } from './TaskManager';
|
||||
|
||||
describe('TaskManager', () => {
|
||||
const logger = getVoidLogger();
|
||||
const databases = TestDatabases.create({
|
||||
ids: ['POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'],
|
||||
});
|
||||
|
||||
async function createDatabase(
|
||||
databaseId: TestDatabaseId,
|
||||
): Promise<DatabaseManager> {
|
||||
const knex = await databases.init(databaseId);
|
||||
const databaseManager: Partial<DatabaseManager> = {
|
||||
forPlugin: () => ({
|
||||
getClient: async () => knex,
|
||||
}),
|
||||
};
|
||||
return databaseManager as DatabaseManager;
|
||||
}
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'can return a working plugin impl, %p',
|
||||
async databaseId => {
|
||||
const database = await createDatabase(databaseId);
|
||||
const manager = new TaskManager(database, logger).forPlugin('test');
|
||||
|
||||
const lock = await manager.acquireLock('lock1', {
|
||||
timeout: Duration.fromMillis(5000),
|
||||
});
|
||||
expect(lock.acquired).toBe(true);
|
||||
},
|
||||
);
|
||||
});
|
||||
@@ -0,0 +1,80 @@
|
||||
/*
|
||||
* 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 } from '@backstage/backend-common';
|
||||
import { Config } from '@backstage/config';
|
||||
import { memoize } from 'lodash';
|
||||
import { Duration } from 'luxon';
|
||||
import { Logger } from 'winston';
|
||||
import { migrateBackendTasks } from '../database/migrateBackendTasks';
|
||||
import { PluginTaskManagerImpl } from './PluginTaskManagerImpl';
|
||||
import { PluginTaskManagerJanitor } from './PluginTaskManagerJanitor';
|
||||
import { PluginTaskManager } from './types';
|
||||
|
||||
/**
|
||||
* Deals with management and locking related to distributed tasks.
|
||||
*
|
||||
* @public
|
||||
*/
|
||||
export class TaskManager {
|
||||
static fromConfig(
|
||||
config: Config,
|
||||
options?: {
|
||||
databaseManager?: DatabaseManager;
|
||||
logger?: Logger;
|
||||
},
|
||||
): TaskManager {
|
||||
const databaseManager =
|
||||
options?.databaseManager ?? DatabaseManager.fromConfig(config);
|
||||
const logger = (options?.logger || getRootLogger()).child({
|
||||
type: 'taskManager',
|
||||
});
|
||||
return new TaskManager(databaseManager, logger);
|
||||
}
|
||||
|
||||
constructor(
|
||||
private readonly databaseManager: DatabaseManager,
|
||||
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();
|
||||
|
||||
await migrateBackendTasks(knex);
|
||||
|
||||
const janitor = new PluginTaskManagerJanitor({
|
||||
knex,
|
||||
waitBetweenRuns: Duration.fromObject({ minutes: 1 }),
|
||||
logger: this.logger,
|
||||
});
|
||||
janitor.start();
|
||||
|
||||
return knex;
|
||||
});
|
||||
|
||||
return new PluginTaskManagerImpl(
|
||||
databaseFactory,
|
||||
this.logger.child({ plugin: pluginId }),
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,257 @@
|
||||
/*
|
||||
* 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 { Duration } from 'luxon';
|
||||
import waitForExpect from 'wait-for-expect';
|
||||
import { migrateBackendTasks } from '../database/migrateBackendTasks';
|
||||
import { DbTasksRow, DB_TASKS_TABLE } from '../database/tables';
|
||||
import { TaskWorker } from './TaskWorker';
|
||||
import { TaskSettingsV1 } from './types';
|
||||
|
||||
describe('TaskWorker', () => {
|
||||
const logger = getVoidLogger();
|
||||
const databases = TestDatabases.create({
|
||||
ids: ['POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'],
|
||||
});
|
||||
|
||||
beforeEach(() => {
|
||||
jest.resetAllMocks();
|
||||
});
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'can run a single task to completion, %p',
|
||||
async databaseId => {
|
||||
const knex = await databases.init(databaseId);
|
||||
await migrateBackendTasks(knex);
|
||||
|
||||
const fn = jest.fn(
|
||||
async () => new Promise<void>(resolve => setTimeout(resolve, 50)),
|
||||
);
|
||||
const settings: TaskSettingsV1 = {
|
||||
version: 1,
|
||||
};
|
||||
|
||||
const worker = new TaskWorker('task1', fn, knex, logger);
|
||||
await worker.start(settings);
|
||||
|
||||
waitForExpect(() => {
|
||||
expect(fn).toBeCalledTimes(1);
|
||||
});
|
||||
},
|
||||
60_000,
|
||||
);
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'goes through the expected states for a single run, %p',
|
||||
async databaseId => {
|
||||
const knex = await databases.init(databaseId);
|
||||
await migrateBackendTasks(knex);
|
||||
|
||||
const fn = jest.fn(
|
||||
async () => new Promise<void>(resolve => setTimeout(resolve, 50)),
|
||||
);
|
||||
const settings: TaskSettingsV1 = {
|
||||
version: 1,
|
||||
initialDelayDuration: Duration.fromObject({ seconds: 1 }).toISO(),
|
||||
recurringAtMostEveryDuration: undefined,
|
||||
timeoutAfterDuration: undefined,
|
||||
};
|
||||
|
||||
const worker = new TaskWorker('task1', fn, knex, logger);
|
||||
await worker.persistTask(settings);
|
||||
|
||||
let row = (await knex<DbTasksRow>(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: 1,
|
||||
initialDelayDuration: 'PT1S',
|
||||
});
|
||||
|
||||
await expect(worker.findReadyTask()).resolves.toEqual({
|
||||
result: 'not ready yet',
|
||||
});
|
||||
|
||||
waitForExpect(async () => {
|
||||
await expect(worker.findReadyTask()).resolves.toEqual({
|
||||
result: 'ready',
|
||||
});
|
||||
});
|
||||
|
||||
row = (await knex<DbTasksRow>(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<DbTasksRow>(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: null,
|
||||
}),
|
||||
);
|
||||
|
||||
await expect(worker.tryReleaseTask('ticket', settings)).resolves.toBe(
|
||||
true,
|
||||
);
|
||||
|
||||
row = (await knex<DbTasksRow>(DB_TASKS_TABLE))[0];
|
||||
expect(row).toBeUndefined();
|
||||
},
|
||||
60_000,
|
||||
);
|
||||
|
||||
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: TaskSettingsV1 = {
|
||||
version: 1,
|
||||
initialDelayDuration: undefined,
|
||||
recurringAtMostEveryDuration: Duration.fromMillis(0).toISO(),
|
||||
timeoutAfterDuration: undefined,
|
||||
};
|
||||
|
||||
const worker = new TaskWorker('task1', fn, knex, logger);
|
||||
worker.start(settings);
|
||||
|
||||
waitForExpect(() => {
|
||||
expect(fn).toBeCalledTimes(3);
|
||||
});
|
||||
},
|
||||
60_000,
|
||||
);
|
||||
|
||||
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<void>(resolve => setTimeout(resolve, 50)),
|
||||
);
|
||||
const settings: TaskSettingsV1 = {
|
||||
version: 1,
|
||||
recurringAtMostEveryDuration: Duration.fromMillis(0).toISO(),
|
||||
};
|
||||
|
||||
const worker = new TaskWorker('task1', fn, knex, logger);
|
||||
await worker.persistTask(settings);
|
||||
await expect(worker.findReadyTask()).resolves.toEqual({
|
||||
result: 'ready',
|
||||
settings,
|
||||
});
|
||||
await expect(worker.tryClaimTask('ticket', settings)).resolves.toBe(true);
|
||||
|
||||
let row = (await knex<DbTasksRow>(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: null,
|
||||
}),
|
||||
);
|
||||
|
||||
await knex<DbTasksRow>(DB_TASKS_TABLE)
|
||||
.where('id', '=', 'task1')
|
||||
.update({ current_run_ticket: 'stolen' });
|
||||
|
||||
await expect(worker.tryReleaseTask('ticket', settings)).resolves.toBe(
|
||||
false,
|
||||
);
|
||||
|
||||
row = (await knex<DbTasksRow>(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: null,
|
||||
}),
|
||||
);
|
||||
},
|
||||
60_000,
|
||||
);
|
||||
|
||||
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: TaskSettingsV1 = {
|
||||
version: 1,
|
||||
recurringAtMostEveryDuration: Duration.fromMillis(0).toISO(),
|
||||
};
|
||||
|
||||
const worker1 = new TaskWorker('task1', fn, knex, logger);
|
||||
await worker1.persistTask(settings);
|
||||
await knex<DbTasksRow>(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 expect(worker2.findReadyTask()).resolves.toEqual({
|
||||
result: 'ready',
|
||||
settings,
|
||||
});
|
||||
await knex<DbTasksRow>(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 expect(worker3.findReadyTask()).resolves.toEqual({
|
||||
result: 'ready',
|
||||
settings,
|
||||
});
|
||||
await expect(worker3.tryClaimTask('ticket', settings)).resolves.toBe(
|
||||
true,
|
||||
);
|
||||
await knex<DbTasksRow>(DB_TASKS_TABLE).where('id', '=', 'task3').delete();
|
||||
await expect(worker3.tryReleaseTask('ticket', settings)).resolves.toBe(
|
||||
false,
|
||||
);
|
||||
},
|
||||
60_000,
|
||||
);
|
||||
});
|
||||
@@ -0,0 +1,277 @@
|
||||
/*
|
||||
* 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 { Knex } from 'knex';
|
||||
import { Duration } from 'luxon';
|
||||
import { v4 as uuid } from 'uuid';
|
||||
import { Logger } from 'winston';
|
||||
import { DbTasksRow, DB_TASKS_TABLE } from '../database/tables';
|
||||
import { CancelToken } from './CancelToken';
|
||||
import { TaskSettingsV1, taskSettingsV1Schema } from './types';
|
||||
import { nowPlus } from './util';
|
||||
|
||||
const WORK_CHECK_FREQUENCY = Duration.fromObject({ seconds: 5 });
|
||||
|
||||
/**
|
||||
* Performs the actual work of a task.
|
||||
*
|
||||
* @private
|
||||
*/
|
||||
export class TaskWorker {
|
||||
private readonly taskId: string;
|
||||
private readonly fn: () => void | Promise<void>;
|
||||
private readonly knex: Knex;
|
||||
private readonly logger: Logger;
|
||||
private readonly cancelToken: CancelToken;
|
||||
|
||||
constructor(
|
||||
taskId: string,
|
||||
fn: () => void | Promise<void>,
|
||||
knex: Knex,
|
||||
logger: Logger,
|
||||
) {
|
||||
this.taskId = taskId;
|
||||
this.fn = fn;
|
||||
this.knex = knex;
|
||||
this.logger = logger;
|
||||
this.cancelToken = CancelToken.create();
|
||||
}
|
||||
|
||||
async start(settings: TaskSettingsV1) {
|
||||
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)}`,
|
||||
);
|
||||
|
||||
(async () => {
|
||||
try {
|
||||
while (!this.cancelToken.isCancelled) {
|
||||
const runResult = await this.runOnce();
|
||||
if (runResult.result === 'abort') {
|
||||
break;
|
||||
}
|
||||
if (!settings.recurringAtMostEveryDuration) {
|
||||
break;
|
||||
}
|
||||
|
||||
await this.sleep(WORK_CHECK_FREQUENCY);
|
||||
}
|
||||
this.logger.info(`Task worker finished: ${this.taskId}`);
|
||||
} catch (e) {
|
||||
this.logger.warn(`Task worker failed unexpectedly, ${e}`);
|
||||
}
|
||||
})();
|
||||
}
|
||||
|
||||
stop() {
|
||||
this.cancelToken.cancel();
|
||||
}
|
||||
|
||||
/**
|
||||
* Makes a single attempt at running the task to completion, if ready.
|
||||
*
|
||||
* @returns The outcome of the attempt
|
||||
*/
|
||||
async runOnce(): 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' };
|
||||
}
|
||||
|
||||
try {
|
||||
await this.fn();
|
||||
} catch (e) {
|
||||
await this.tryReleaseTask(ticket, taskSettings);
|
||||
return { result: 'failed' };
|
||||
}
|
||||
|
||||
await this.tryReleaseTask(ticket, taskSettings);
|
||||
return { result: 'completed' };
|
||||
}
|
||||
|
||||
/**
|
||||
* Sleep for the given duration, but abort sooner if the cancel token
|
||||
* triggers.
|
||||
*
|
||||
* @param duration - The amount of time to sleep, at most
|
||||
*/
|
||||
private async sleep(duration: Duration): Promise<void> {
|
||||
await Promise.race([
|
||||
new Promise(resolve => setTimeout(resolve, duration.as('milliseconds'))),
|
||||
this.cancelToken.promise,
|
||||
]);
|
||||
}
|
||||
|
||||
/**
|
||||
* Perform the initial store of the task info
|
||||
*/
|
||||
async persistTask(settings: TaskSettingsV1) {
|
||||
// Perform an initial parse to ensure that we will definitely be able to
|
||||
// read it back again.
|
||||
taskSettingsV1Schema.parse(settings);
|
||||
|
||||
const settingsJson = JSON.stringify(settings);
|
||||
const startAt = settings.initialDelayDuration
|
||||
? nowPlus(Duration.fromISO(settings.initialDelayDuration), this.knex)
|
||||
: this.knex.fn.now();
|
||||
|
||||
// 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.
|
||||
await this.knex<DbTasksRow>(DB_TASKS_TABLE)
|
||||
.insert({
|
||||
id: this.taskId,
|
||||
settings_json: settingsJson,
|
||||
next_run_start_at: startAt,
|
||||
})
|
||||
.onConflict('id')
|
||||
.merge(['settings_json']);
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if the task is ready to run
|
||||
*/
|
||||
async findReadyTask(): Promise<
|
||||
| { result: 'not ready yet' }
|
||||
| { result: 'abort' }
|
||||
| { result: 'ready'; settings: TaskSettingsV1 }
|
||||
> {
|
||||
const [row] = await this.knex<DbTasksRow>(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 settings = taskSettingsV1Schema.parse(JSON.parse(row.settingsJson));
|
||||
return { result: 'ready', settings };
|
||||
} catch (e) {
|
||||
this.logger.info(
|
||||
'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: TaskSettingsV1,
|
||||
): Promise<boolean> {
|
||||
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<DbTasksRow>(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: TaskSettingsV1,
|
||||
): Promise<boolean> {
|
||||
const { recurringAtMostEveryDuration } = settings;
|
||||
|
||||
// If this is not a recurring task, and we still have the current run
|
||||
// ticket, delete it from the table
|
||||
if (recurringAtMostEveryDuration === undefined) {
|
||||
const rows = await this.knex<DbTasksRow>(DB_TASKS_TABLE)
|
||||
.where('id', '=', this.taskId)
|
||||
.where('current_run_ticket', '=', ticket)
|
||||
.delete();
|
||||
|
||||
return rows === 1;
|
||||
}
|
||||
|
||||
// We make an effort to keep the datetime calculations in the database
|
||||
// layer, making sure to not have to perform conversions back and forth and
|
||||
// leaning on the database as a central clock source
|
||||
const dbNull = this.knex.raw('null');
|
||||
const dt = Duration.fromISO(recurringAtMostEveryDuration).as('seconds');
|
||||
const nextRun =
|
||||
this.knex.client.config.client === 'sqlite3'
|
||||
? this.knex.raw('datetime(next_run_start_at, ?)', [`+${dt} seconds`])
|
||||
: this.knex.raw(`next_run_start_at + interval '${dt} seconds'`);
|
||||
|
||||
const rows = await this.knex<DbTasksRow>(DB_TASKS_TABLE)
|
||||
.where('id', '=', this.taskId)
|
||||
.where('current_run_ticket', '=', ticket)
|
||||
.update({
|
||||
next_run_start_at: nextRun,
|
||||
current_run_ticket: dbNull,
|
||||
current_run_started_at: dbNull,
|
||||
current_run_expires_at: dbNull,
|
||||
});
|
||||
|
||||
return rows === 1;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
/*
|
||||
* 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 { TaskManager } from './TaskManager';
|
||||
export type { LockOptions, PluginTaskManager, TaskOptions } from './types';
|
||||
@@ -0,0 +1,154 @@
|
||||
/*
|
||||
* 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 { 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.
|
||||
*
|
||||
* @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: LockOptions,
|
||||
): Promise<
|
||||
{ acquired: false } | { acquired: true; release(): Promise<void> }
|
||||
>;
|
||||
|
||||
/**
|
||||
* 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: TaskOptions,
|
||||
fn: () => void | Promise<void>,
|
||||
): Promise<{ unschedule: () => Promise<void> }>;
|
||||
}
|
||||
|
||||
function isValidOptionalDurationString(d: string | undefined): boolean {
|
||||
try {
|
||||
return !d || Duration.fromISO(d).isValid === true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
export const taskSettingsV1Schema = z.object({
|
||||
version: z.literal(1),
|
||||
initialDelayDuration: z
|
||||
.string()
|
||||
.optional()
|
||||
.refine(isValidOptionalDurationString, { message: 'Invalid duration' }),
|
||||
recurringAtMostEveryDuration: z
|
||||
.string()
|
||||
.optional()
|
||||
.refine(isValidOptionalDurationString, { message: 'Invalid duration' }),
|
||||
timeoutAfterDuration: z
|
||||
.string()
|
||||
.optional()
|
||||
.refine(isValidOptionalDurationString, { message: 'Invalid duration' }),
|
||||
});
|
||||
|
||||
/**
|
||||
* The properties that control a scheduled task (version 1).
|
||||
*/
|
||||
export type TaskSettingsV1 = z.infer<typeof taskSettingsV1Schema>;
|
||||
@@ -0,0 +1,45 @@
|
||||
/*
|
||||
* 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
|
||||
export function validateId(id: string) {
|
||||
if (typeof id !== 'string' || !/^[a-z0-9]+(?:_[a-z0-9]+)*$/.test(id)) {
|
||||
throw new InputError(
|
||||
`${id} is not a valid ID, expected string of lowercase characters and digits separated by underscores`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
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();
|
||||
}
|
||||
return knex.client.config.client === 'sqlite3'
|
||||
? knex.raw(`datetime('now', ?)`, [`${seconds} seconds`])
|
||||
: knex.raw(`now() + interval '${seconds} seconds'`);
|
||||
}
|
||||
Reference in New Issue
Block a user