paulononato/backstage
1
0
Fork 0
Code Issues Pull Requests Actions 30 Packages Projects Releases Wiki Activity

Merge pull request #7612 from backstage/freben/recurring-tasks

Browse Source 
Add support for distributed tasks
This commit is contained in:
Fredrik Adelöw
2021-11-12 14:32:30 +01:00
committed by GitHub
parent 7f839ddbb6 5f606617dd
commit b46cd73035
32 changed files with 1544 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/backend-common/api-report.md
+3
View File
@@ -374,6 +374,9 @@ export class GitlabUrlReader implements UrlReader {
export { isChildPath };
// @public
export function isDatabaseConflictError(e: unknown): boolean;
// @public
export function loadBackendConfig(options: {
logger: Logger_2;
packages/backend-common/src/database/DatabaseManager.ts
+3 -2
View File
@@ -13,14 +13,15 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import { Knex } from 'knex';
import { omit } from 'lodash';
import { Config, ConfigReader } from '@backstage/config';
import { JsonObject } from '@backstage/types';
import {
createDatabaseClient,
ensureDatabaseExists,
createNameOverride,
ensureDatabaseExists,
normalizeConnection,
} from './connection';
import { PluginDatabaseManager } from './types';
@@ -165,7 +166,7 @@ export class DatabaseManager {
);
return {
// include base connection if client type has not been overriden
// include base connection if client type has not been overridden
...(overridden ? {} : baseConnection),
...connection,
};
packages/backend-common/src/database/index.ts
+1
View File
@@ -28,3 +28,4 @@ export {
} from './connection';
export type { PluginDatabaseManager } from './types';
export { isDatabaseConflictError } from './util';
packages/backend-common/src/database/util.ts
+33
View File
@@ -0,0 +1,33 @@
/*
* 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.
*/
/**
* Tries to deduce whether a thrown error is a database conflict.
*
* @public
* @param e - A thrown error
* @returns True if the error looks like it was a conflict error thrown by a
* known database engine
*/
export function isDatabaseConflictError(e: unknown) {
const message = (e as any)?.message;
return (
typeof message === 'string' &&
(/SQLITE_CONSTRAINT: UNIQUE/.test(message) ||
/unique constraint/.test(message))
);
}
packages/backend-tasks/.eslintrc.js
+3
View File
@@ -0,0 +1,3 @@
module.exports = {
extends: [require.resolve('@backstage/cli/config/eslint.backend')],
};
packages/backend-tasks/README.md
+35
View File
@@ -0,0 +1,35 @@
# @backstage/backend-tasks
Common distributed task management 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 { TaskScheduler } from '@backstage/backend-tasks';
import { Duration } from 'luxon';
const scheduler = TaskScheduler.fromConfig(rootConfig).forPlugin('my-plugin');
await scheduler.scheduleTask({
id: 'refresh-things',
frequency: Duration.fromObject({ minutes: 10 }),
fn: 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)
packages/backend-tasks/api-report.md
+45
View File
@@ -0,0 +1,45 @@
## 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 { AbortSignal as AbortSignal_2 } from 'node-abort-controller';
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 PluginTaskScheduler {
scheduleTask(task: TaskDefinition): Promise<void>;
}
// @public
export interface TaskDefinition {
fn: TaskFunction;
frequency: Duration;
id: string;
initialDelay?: Duration;
signal?: AbortSignal_2;
timeout: Duration;
}
// @public
export type TaskFunction =
| ((abortSignal: AbortSignal_2) => void | Promise<void>)
| (() => void | Promise<void>);
// @public
export class TaskScheduler {
constructor(databaseManager: DatabaseManager, logger: Logger_2);
forPlugin(pluginId: string): PluginTaskScheduler;
// (undocumented)
static fromConfig(
config: Config,
options?: {
databaseManager?: DatabaseManager;
logger?: Logger_2;
},
): TaskScheduler;
}
```
packages/backend-tasks/migrations/20210928160613_init.js
+64
View File
@@ -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
.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')
.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');
};
packages/backend-tasks/package.json
+55
View File
@@ -0,0 +1,55 @@
{
"name": "@backstage/backend-tasks",
"description": "Common distributed task management 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.8",
"@backstage/config": "^0.1.11",
"@backstage/errors": "^0.1.4",
"@backstage/types": "^0.1.1",
"@types/luxon": "^2.0.4",
"knex": "^0.95.1",
"lodash": "^4.17.21",
"luxon": "^2.0.2",
"node-abort-controller": "^3.0.1",
"uuid": "^8.0.0",
"winston": "^3.2.1",
"zod": "^3.9.5"
},
"devDependencies": {
"@backstage/backend-test-utils": "^0.1.8",
"@backstage/cli": "^0.8.1",
"jest": "^26.0.1",
"wait-for-expect": "^3.0.2"
},
"files": [
"dist",
"migrations/**/*.{js,d.ts}"
]
}
packages/backend-tasks/src/database/migrateBackendTasks.ts
+31
View File
@@ -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,
});
}
packages/backend-tasks/src/database/tables.ts
+27
View File
@@ -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;
};
packages/backend-tasks/src/index.ts
+23
View File
@@ -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 library for Backstage backends
*
* @packageDocumentation
*/
export * from './tasks';
packages/backend-tasks/src/setupTests.ts
+17
View File
@@ -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 {};
packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.test.ts
+62
View File
@@ -0,0 +1,62 @@
/*
* 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 { PluginTaskSchedulerImpl } from './PluginTaskSchedulerImpl';
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 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', () => {
it.each(databases.eachSupportedId())(
'can run the happy path, %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,
});
await waitForExpect(() => {
expect(fn).toBeCalled();
});
},
60_000,
);
});
});
packages/backend-tasks/src/tasks/PluginTaskSchedulerImpl.ts
+50
View File
@@ -0,0 +1,50 @@
/*
* 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 { Logger } from 'winston';
import { TaskWorker } from './TaskWorker';
import { PluginTaskScheduler, TaskDefinition } from './types';
import { validateId } from './util';
/**
* Implements the actual task management.
*/
export class PluginTaskSchedulerImpl implements PluginTaskScheduler {
constructor(
private readonly databaseFactory: () => Promise<Knex>,
private readonly logger: Logger,
) {}
async scheduleTask(task: TaskDefinition): Promise<void> {
validateId(task.id);
const knex = await this.databaseFactory();
const worker = new TaskWorker(task.id, task.fn, knex, this.logger);
await worker.start(
{
version: 1,
initialDelayDuration: task.initialDelay?.toISO(),
recurringAtMostEveryDuration: task.frequency.toISO(),
timeoutAfterDuration: task.timeout.toISO(),
},
{
signal: task.signal,
},
);
}
}
packages/backend-tasks/src/tasks/PluginTaskSchedulerJanitor.ts
+83
View File
@@ -0,0 +1,83 @@
/*
* 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 { AbortSignal } from 'node-abort-controller';
import { Logger } from 'winston';
import { DbTasksRow, DB_TASKS_TABLE } 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: Logger;
constructor(options: {
knex: Knex;
waitBetweenRuns: Duration;
logger: Logger;
}) {
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() {
// 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 dbNull = this.knex.raw('null');
const tasksQuery = this.knex<DbTasksRow>(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,
});
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}`);
}
}
}
}
packages/backend-tasks/src/tasks/TaskScheduler.test.ts
+61
View File
@@ -0,0 +1,61 @@
/*
* 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 { TaskScheduler } from './TaskScheduler';
import waitForExpect from 'wait-for-expect';
describe('TaskScheduler', () => {
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 TaskScheduler(database, logger).forPlugin('test');
const fn = jest.fn();
await manager.scheduleTask({
id: 'task1',
timeout: Duration.fromMillis(5000),
frequency: Duration.fromMillis(5000),
fn,
});
await waitForExpect(() => {
expect(fn).toBeCalled();
});
},
60_000,
);
});
packages/backend-tasks/src/tasks/TaskScheduler.ts
+80
View File
@@ -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 { once } from 'lodash';
import { Duration } from 'luxon';
import { Logger } from 'winston';
import { migrateBackendTasks } from '../database/migrateBackendTasks';
import { PluginTaskSchedulerImpl } from './PluginTaskSchedulerImpl';
import { PluginTaskSchedulerJanitor } from './PluginTaskSchedulerJanitor';
import { PluginTaskScheduler } from './types';
/**
* Deals with the scheduling of distributed tasks.
*
* @public
*/
export class TaskScheduler {
static fromConfig(
config: Config,
options?: {
databaseManager?: DatabaseManager;
logger?: Logger;
},
): 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: 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 PluginTaskScheduler} instance
*/
forPlugin(pluginId: string): PluginTaskScheduler {
const databaseFactory = once(async () => {
const knex = await this.databaseManager.forPlugin(pluginId).getClient();
await migrateBackendTasks(knex);
const janitor = new PluginTaskSchedulerJanitor({
knex,
waitBetweenRuns: Duration.fromObject({ minutes: 1 }),
logger: this.logger,
});
janitor.start();
return knex;
});
return new PluginTaskSchedulerImpl(
databaseFactory,
this.logger.child({ plugin: pluginId }),
);
}
}
packages/backend-tasks/src/tasks/TaskWorker.test.ts
+245
View File
@@ -0,0 +1,245 @@
/*
* 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())(
'goes through the expected states, %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.fromMillis(1000).toISO(),
recurringAtMostEveryDuration: Duration.fromMillis(2000).toISO(),
timeoutAfterDuration: Duration.fromMillis(60000).toISO(),
};
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',
recurringAtMostEveryDuration: 'PT2S',
timeoutAfterDuration: 'PT60S',
});
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: expect.anything(),
}),
);
await expect(worker.tryReleaseTask('ticket', settings)).resolves.toBe(
true,
);
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,
}),
);
},
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: Duration.fromMillis(60000).toISO(),
};
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(),
timeoutAfterDuration: Duration.fromMillis(60000).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: expect.anything(),
}),
);
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: expect.anything(),
}),
);
},
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(),
timeoutAfterDuration: Duration.fromMillis(60000).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,
);
});
packages/backend-tasks/src/tasks/TaskWorker.ts
+249
View File
@@ -0,0 +1,249 @@
/*
* 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 { AbortSignal } from 'node-abort-controller';
import { v4 as uuid } from 'uuid';
import { Logger } from 'winston';
import { DbTasksRow, DB_TASKS_TABLE } from '../database/tables';
import { TaskFunction, TaskSettingsV1, taskSettingsV1Schema } from './types';
import { delegateAbortController, nowPlus, sleep } 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: TaskFunction;
private readonly knex: Knex;
private readonly logger: Logger;
constructor(taskId: string, fn: TaskFunction, knex: Knex, logger: Logger) {
this.taskId = taskId;
this.fn = fn;
this.knex = knex;
this.logger = logger;
}
async start(settings: TaskSettingsV1, 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)}`,
);
(async () => {
try {
while (!options?.signal?.aborted) {
const runResult = await this.runOnce(options?.signal);
if (runResult.result === 'abort') {
break;
}
await sleep(WORK_CHECK_FREQUENCY, options?.signal);
}
this.logger.info(`Task worker finished: ${this.taskId}`);
} catch (e) {
this.logger.warn(`Task worker failed unexpectedly, ${e}`);
}
})();
}
/**
* Makes a single attempt at running the task to completion, if ready.
*
* @returns The outcome of the attempt
*/
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);
} catch (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: 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(
`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: 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;
// 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;
}
}
packages/backend-tasks/src/tasks/index.ts
+22
View File
@@ -0,0 +1,22 @@
/*
* 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 { TaskScheduler } from './TaskScheduler';
export type {
PluginTaskScheduler,
TaskDefinition,
TaskFunction,
} from './types';
packages/backend-tasks/src/tasks/types.ts
+138
View File
@@ -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 { Duration } from 'luxon';
import { AbortSignal } from 'node-abort-controller';
import { z } from 'zod';
/**
* A function that can be called as a scheduled task.
*
* It may optionally accept an abort signal argument. When the signal triggers,
* processing should abort and return as quickly as possible.
*
* @public
*/
export type TaskFunction =
| ((abortSignal: AbortSignal) => void | Promise<void>)
| (() => void | Promise<void>);
/**
* Options that apply to the invocation of a given task.
*
* @public
*/
export interface TaskDefinition {
/**
* A unique ID (within the scope of the plugin) for the task.
*/
id: string;
/**
* The actual task function to be invoked regularly.
*/
fn: TaskFunction;
/**
* An abort signal that, when triggered, will stop the recurring execution of
* the task.
*/
signal?: AbortSignal;
/**
* 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 the scheduling of distributed tasks, for a given plugin.
*
* @public
*/
export interface PluginTaskScheduler {
/**
* 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 definition - The task definition
*/
scheduleTask(task: TaskDefinition): 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()
.refine(isValidOptionalDurationString, { message: 'Invalid duration' }),
timeoutAfterDuration: z
.string()
.refine(isValidOptionalDurationString, { message: 'Invalid duration' }),
});
/**
* The properties that control a scheduled task (version 1).
*/
export type TaskSettingsV1 = z.infer<typeof taskSettingsV1Schema>;
packages/backend-tasks/src/tasks/util.test.ts
+76
View File
@@ -0,0 +1,76 @@
/*
* 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 { AbortController } from 'node-abort-controller';
import { delegateAbortController, sleep, validateId } from './util';
describe('util', () => {
describe('validateId', () => {
it.each(['a', 'a_b', 'ab123c_2'])(
'accepts valid inputs, %p',
async input => {
expect(validateId(input)).toBeUndefined();
},
);
it.each(['', 'a!', 'A', 'a-b', 'a.b', '_a', 'a_', 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);
});
});
});
packages/backend-tasks/src/tasks/util.ts
+107
View File
@@ -0,0 +1,107 @@
/*
* 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';
import { AbortController, AbortSignal } from 'node-abort-controller';
// 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'`);
}
/**
* 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<void> {
if (abortSignal?.aborted) {
return;
}
await new Promise<void>(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;
}
packages/backend/src/index.ts
+1 -1
View File
@@ -48,12 +48,12 @@ import scaffolder from './plugins/scaffolder';
import proxy from './plugins/proxy';
import search from './plugins/search';
import techdocs from './plugins/techdocs';
import techInsights from './plugins/techInsights';
import todo from './plugins/todo';
import graphql from './plugins/graphql';
import app from './plugins/app';
import badges from './plugins/badges';
import jenkins from './plugins/jenkins';
import techInsights from './plugins/techInsights';
import { PluginEnvironment } from './types';
function makeCreateEnv(config: Config) {
packages/catalog-client/package.json
+1 -1
View File
@@ -31,7 +31,7 @@
},
"dependencies": {
"@backstage/catalog-model": "^0.9.5",
"@backstage/errors": "^0.1.3",
"@backstage/errors": "^0.1.4",
"cross-fetch": "^3.0.6"
},
"devDependencies": {
packages/e2e-test/package.json
+1 -1
View File
@@ -25,7 +25,7 @@
},
"devDependencies": {
"@backstage/cli-common": "^0.1.1",
"@backstage/errors": "^0.1.2",
"@backstage/errors": "^0.1.4",
"@types/fs-extra": "^9.0.1",
"@types/node": "^14.14.32",
"@types/puppeteer": "^5.4.4",