From d088c7f0a3d98ead6101f27d1c017e639e0c3c9d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Mon, 23 Aug 2021 15:29:03 +0200 Subject: [PATCH] Implement the locks part of the task manager MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- packages/backend-common/knexfile.js | 26 ++++ .../migrations/20210928160613_init.js | 71 +++++++++ packages/backend-common/package.json | 5 + .../src/database/DatabaseManager.ts | 5 +- .../src/database/migrateBackendCommon.ts | 30 ++++ .../backend-common/src/database/tables.ts | 22 +++ packages/backend-common/src/index.ts | 1 + .../src/tasks/TaskManager.test.ts | 73 +++++++++ .../backend-common/src/tasks/TaskManager.ts | 144 ++++++++++++++++++ packages/backend-common/src/tasks/index.ts | 18 +++ packages/backend-common/src/tasks/types.ts | 45 ++++++ packages/backend-common/src/tasks/util.ts | 26 ++++ packages/backend/src/index.ts | 31 ++-- packages/backend/src/types.ts | 6 +- yarn.lock | 5 + 15 files changed, 490 insertions(+), 18 deletions(-) create mode 100644 packages/backend-common/knexfile.js create mode 100644 packages/backend-common/migrations/20210928160613_init.js create mode 100644 packages/backend-common/src/database/migrateBackendCommon.ts create mode 100644 packages/backend-common/src/database/tables.ts create mode 100644 packages/backend-common/src/tasks/TaskManager.test.ts create mode 100644 packages/backend-common/src/tasks/TaskManager.ts create mode 100644 packages/backend-common/src/tasks/index.ts create mode 100644 packages/backend-common/src/tasks/types.ts create mode 100644 packages/backend-common/src/tasks/util.ts diff --git a/packages/backend-common/knexfile.js b/packages/backend-common/knexfile.js new file mode 100644 index 0000000000..4c8be42673 --- /dev/null +++ b/packages/backend-common/knexfile.js @@ -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', + }, +}; diff --git a/packages/backend-common/migrations/20210928160613_init.js b/packages/backend-common/migrations/20210928160613_init.js new file mode 100644 index 0000000000..663e2f8aeb --- /dev/null +++ b/packages/backend-common/migrations/20210928160613_init.js @@ -0,0 +1,71 @@ +/* + * 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) { + // + // locking + // + await knex.schema.createTable( + 'backstage_backend_common__task_locks', + table => { + table.comment('Locks used for mutual exclusion among multiple workers'); + table + .text('id') + .primary() + .notNullable() + .comment('The unique id of this particular lock'); + table + .text('acquired_ticket') + .nullable() + .comment('A unique ticket for the current lock acquiral, if any'); + table + .dateTime('acquired_at') + .nullable() + .comment('The time when the lock was acquired, if locked'); + table + .dateTime('expires_at') + .nullable() + .comment('The time when an acquired lock will time out and expire'); + table.index('id', 'task_locks_id_idx'); + }, + ); + // + // tasks + // + await knex.schema.createTable('backstage_backend_common__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'); + }); +}; + +/** + * @param {import('knex').Knex} knex + */ +exports.down = async function down(knex) { + await knex.schema.alterTable('task_locks', table => { + table.dropIndex([], 'task_locks_id_idx'); + }); + await knex.schema.dropTable('task_locks'); +}; diff --git a/packages/backend-common/package.json b/packages/backend-common/package.json index 03ae1fbe62..c8e8d69ab5 100644 --- a/packages/backend-common/package.json +++ b/packages/backend-common/package.json @@ -59,14 +59,17 @@ "knex": "^0.95.1", "lodash": "^4.17.21", "logform": "^2.1.1", + "luxon": "^2.0.2", "minimatch": "^3.0.4", "minimist": "^1.2.5", "morgan": "^1.10.0", + "node-abort-controller": "^3.0.0", "raw-body": "^2.4.1", "selfsigned": "^1.10.7", "stoppable": "^1.1.0", "tar": "^6.1.2", "unzipper": "^0.10.11", + "uuid": "^8.0.0", "winston": "^3.2.1", "yn": "^4.0.0" }, @@ -79,6 +82,7 @@ } }, "devDependencies": { + "@backstage/backend-test-utils": "^0.1.8", "@backstage/cli": "^0.8.2", "@backstage/test-utils": "^0.1.21", "@types/archiver": "^5.1.0", @@ -107,6 +111,7 @@ }, "files": [ "dist", + "migrations/**/*.{js,d.ts}", "config.d.ts" ], "configSchema": "config.d.ts" diff --git a/packages/backend-common/src/database/DatabaseManager.ts b/packages/backend-common/src/database/DatabaseManager.ts index c0511c6b78..f8cdab1413 100644 --- a/packages/backend-common/src/database/DatabaseManager.ts +++ b/packages/backend-common/src/database/DatabaseManager.ts @@ -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, }; diff --git a/packages/backend-common/src/database/migrateBackendCommon.ts b/packages/backend-common/src/database/migrateBackendCommon.ts new file mode 100644 index 0000000000..6283f3a96c --- /dev/null +++ b/packages/backend-common/src/database/migrateBackendCommon.ts @@ -0,0 +1,30 @@ +/* + * 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 { resolvePackagePath } from '../paths'; + +const migrationsDir = resolvePackagePath( + '@backstage/backend-common', + 'migrations', +); + +export async function migrateBackendCommon(knex: Knex): Promise { + await knex.migrate.latest({ + directory: migrationsDir, + tableName: 'knex_migrations_backstage_backend_common', + }); +} diff --git a/packages/backend-common/src/database/tables.ts b/packages/backend-common/src/database/tables.ts new file mode 100644 index 0000000000..8efa393bb7 --- /dev/null +++ b/packages/backend-common/src/database/tables.ts @@ -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 type DbTaskLocksRow = { + id: string; + acquired_ticket?: string; + acquired_at?: Date; + expires_at?: Date; +}; diff --git a/packages/backend-common/src/index.ts b/packages/backend-common/src/index.ts index c214961a2e..a906d7bc61 100644 --- a/packages/backend-common/src/index.ts +++ b/packages/backend-common/src/index.ts @@ -31,4 +31,5 @@ export * from './paths'; export * from './reading'; export * from './scm'; export * from './service'; +export * from './tasks'; export * from './util'; diff --git a/packages/backend-common/src/tasks/TaskManager.test.ts b/packages/backend-common/src/tasks/TaskManager.test.ts new file mode 100644 index 0000000000..da9adb70f6 --- /dev/null +++ b/packages/backend-common/src/tasks/TaskManager.test.ts @@ -0,0 +1,73 @@ +/* + * 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 { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils'; +import { Duration } from 'luxon'; +import { DatabaseManager } from '../database'; +import { TaskManager } from './TaskManager'; + +describe('TaskManager', () => { + const databases = TestDatabases.create({ + ids: ['POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'], + }); + + async function createDatabase( + databaseId: TestDatabaseId, + ): Promise { + const knex = await databases.init(databaseId); + const databaseManager: Partial = { + forPlugin: () => ({ + getClient: async () => knex, + }), + }; + return databaseManager as DatabaseManager; + } + + describe('locking', () => { + it.each(databases.eachSupportedId())( + 'can run the happy path, %p', + async databaseId => { + const database = await createDatabase(databaseId); + const manager = new TaskManager(database).forPlugin('test'); + + 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(); + }, + ); + }); +}); diff --git a/packages/backend-common/src/tasks/TaskManager.ts b/packages/backend-common/src/tasks/TaskManager.ts new file mode 100644 index 0000000000..29741e1594 --- /dev/null +++ b/packages/backend-common/src/tasks/TaskManager.ts @@ -0,0 +1,144 @@ +/* + * 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 { Config } from '@backstage/config'; +import { Knex } from 'knex'; +import { memoize } from 'lodash'; +import { Duration } from 'luxon'; +import { v4 as uuid } from 'uuid'; +import { DatabaseManager } from '../database'; +import { migrateBackendCommon } from '../database/migrateBackendCommon'; +import { DbTaskLocksRow } from '../database/tables'; +import { PluginTaskManager } from './types'; +import { validateId } from './util'; + +export class PluginTaskManagerImpl implements PluginTaskManager { + constructor( + private readonly pluginId: string, + private readonly databaseFactory: () => Promise, + ) {} + + async acquireLock( + idWithoutPrefix: string, + options: { + timeout: Duration; + }, + ): Promise< + | { acquired: false } + | { acquired: true; release: () => void | Promise } + > { + validateId(idWithoutPrefix); + + const knex = await this.databaseFactory(); + const id = `plugin:${this.pluginId}:${idWithoutPrefix}`; + const ticket = uuid(); + const timeout = options.timeout.as('seconds'); + + const release = async () => { + try { + await knex('backstage_backend_common__task_locks') + .where('id', '=', id) + .where('acquired_ticket', '=', ticket) + .delete(); + } catch { + // fail silently + } + }; + + // First try to overwrite an existing lock, that has timed out + const stolen = await knex( + 'backstage_backend_common__task_locks', + ) + .where('id', '=', id) + .whereNotNull('acquired_ticket') + .where('expires_at', '<', knex.fn.now()) + .update({ + acquired_ticket: ticket, + acquired_at: knex.fn.now(), + expires_at: + knex.client.config.client === 'sqlite3' + ? knex.raw(`datetime('now', ?)`, [`${timeout} seconds`]) + : knex.raw(`now() + interval '${timeout} seconds'`), + }); + + if (stolen) { + return { acquired: true, release }; + } + + try { + await knex('backstage_backend_common__task_locks').insert( + { + id, + acquired_ticket: ticket, + acquired_at: knex.fn.now(), + expires_at: + knex.client.config.client === 'sqlite3' + ? knex.raw(`datetime('now', ?)`, [`${timeout} seconds`]) + : knex.raw(`now() + interval '${timeout} seconds'`), + }, + ); + return { acquired: true, release }; + } catch { + return { acquired: false }; + } + } + + async scheduleTask( + idWithoutPrefix: string, + options: { + timeout: Duration; + frequency: Duration; + initialDelay?: Duration; + }, + fn: () => Promise, + ): Promise<{ release: () => Promise }> { + validateId(idWithoutPrefix); + + const knex = await this.databaseFactory(); + const id = `plugin:${this.pluginId}:${idWithoutPrefix}`; + + return {}; + } +} + +/** + * Deals with management and locking related to distributed tasks. + * + * @public + */ +export class TaskManager { + static fromConfig( + config: Config, + options?: { databaseManager?: DatabaseManager }, + ): TaskManager { + const databaseManager = + options?.databaseManager ?? DatabaseManager.fromConfig(config); + return new TaskManager(databaseManager); + } + + constructor(private readonly databaseManager: DatabaseManager) {} + + forPlugin(pluginId: string): PluginTaskManager { + return new PluginTaskManagerImpl( + pluginId, + memoize(async () => { + const knex = await this.databaseManager.forPlugin(pluginId).getClient(); + await migrateBackendCommon(knex); + return knex; + }), + ); + } +} diff --git a/packages/backend-common/src/tasks/index.ts b/packages/backend-common/src/tasks/index.ts new file mode 100644 index 0000000000..60329788b3 --- /dev/null +++ b/packages/backend-common/src/tasks/index.ts @@ -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 type { PluginTaskManager } from './types'; +export { TaskManager } from './TaskManager'; diff --git a/packages/backend-common/src/tasks/types.ts b/packages/backend-common/src/tasks/types.ts new file mode 100644 index 0000000000..4a4fcb474e --- /dev/null +++ b/packages/backend-common/src/tasks/types.ts @@ -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 { Duration } from 'luxon'; + +/** + * Deals with management and locking related to distributed tasks, for a given + * plugin. + * + * @public + */ +export interface PluginTaskManager { + acquireLock( + id: string, + options: { + timeout: Duration; + }, + ): Promise< + | { acquired: false } + | { acquired: true; release: () => void | Promise } + >; + + scheduleTask( + id: string, + options: { + timeout: Duration; + frequency: Duration; + initialDelay?: Duration; + }, + fn: () => Promise, + ): Promise<{ release: () => Promise }>; +} diff --git a/packages/backend-common/src/tasks/util.ts b/packages/backend-common/src/tasks/util.ts new file mode 100644 index 0000000000..7bd4ff61b5 --- /dev/null +++ b/packages/backend-common/src/tasks/util.ts @@ -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. + */ + +import { InputError } from '@backstage/errors'; + +// 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`, + ); + } +} diff --git a/packages/backend/src/index.ts b/packages/backend/src/index.ts index ffdce949b7..554c9a7d99 100644 --- a/packages/backend/src/index.ts +++ b/packages/backend/src/index.ts @@ -22,41 +22,42 @@ * Happy hacking! */ -import Router from 'express-promise-router'; import { CacheManager, createServiceBuilder, + DatabaseManager, getRootLogger, loadBackendConfig, notFoundHandler, - DatabaseManager, SingleHostDiscovery, + TaskManager, UrlReaders, useHotMemoize, } from '@backstage/backend-common'; import { Config } from '@backstage/config'; -import healthcheck from './plugins/healthcheck'; -import { metricsInit, metricsHandler } from './metrics'; +import Router from 'express-promise-router'; +import { metricsHandler, metricsInit } from './metrics'; +import app from './plugins/app'; import auth from './plugins/auth'; import azureDevOps from './plugins/azure-devops'; +import badges from './plugins/badges'; import catalog from './plugins/catalog'; import codeCoverage from './plugins/codecoverage'; -import kubernetes from './plugins/kubernetes'; +import graphql from './plugins/graphql'; +import healthcheck from './plugins/healthcheck'; +import jenkins from './plugins/jenkins'; import kafka from './plugins/kafka'; +import kubernetes from './plugins/kubernetes'; +import proxy from './plugins/proxy'; import rollbar from './plugins/rollbar'; import scaffolder from './plugins/scaffolder'; -import proxy from './plugins/proxy'; import search from './plugins/search'; import techdocs from './plugins/techdocs'; -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 todo from './plugins/todo'; import { PluginEnvironment } from './types'; -function makeCreateEnv(config: Config) { +async function makeCreateEnv(config: Config) { const root = getRootLogger(); const reader = UrlReaders.default({ logger: root, config }); const discovery = SingleHostDiscovery.fromConfig(config); @@ -64,13 +65,15 @@ function makeCreateEnv(config: Config) { root.info(`Created UrlReader ${reader}`); const databaseManager = DatabaseManager.fromConfig(config); + const taskManager = TaskManager.fromConfig(config); const cacheManager = CacheManager.fromConfig(config); return (plugin: string): PluginEnvironment => { const logger = root.child({ type: 'plugin', plugin }); const database = databaseManager.forPlugin(plugin); + const tasks = taskManager.forPlugin(plugin); const cache = cacheManager.forPlugin(plugin); - return { logger, cache, database, config, reader, discovery }; + return { logger, cache, database, tasks, config, reader, discovery }; }; } @@ -87,7 +90,7 @@ async function main() { argv: process.argv, logger, }); - const createEnv = makeCreateEnv(config); + const createEnv = await makeCreateEnv(config); const healthcheckEnv = useHotMemoize(module, () => createEnv('healthcheck')); const catalogEnv = useHotMemoize(module, () => createEnv('catalog')); diff --git a/packages/backend/src/types.ts b/packages/backend/src/types.ts index 8290e569ef..aafd25bc3f 100644 --- a/packages/backend/src/types.ts +++ b/packages/backend/src/types.ts @@ -14,19 +14,21 @@ * limitations under the License. */ -import { Logger } from 'winston'; -import { Config } from '@backstage/config'; import { PluginCacheManager, PluginDatabaseManager, PluginEndpointDiscovery, + PluginTaskManager, UrlReader, } from '@backstage/backend-common'; +import { Config } from '@backstage/config'; +import { Logger } from 'winston'; export type PluginEnvironment = { logger: Logger; cache: PluginCacheManager; database: PluginDatabaseManager; + tasks: PluginTaskManager; config: Config; reader: UrlReader; discovery: PluginEndpointDiscovery; diff --git a/yarn.lock b/yarn.lock index 4c42941ad9..caba3db5fb 100644 --- a/yarn.lock +++ b/yarn.lock @@ -20895,6 +20895,11 @@ node-abi@^2.21.0: dependencies: semver "^5.4.1" +node-abort-controller@^3.0.0: + version "3.0.0" + resolved "https://registry.npmjs.org/node-abort-controller/-/node-abort-controller-3.0.0.tgz#e7b1f2b72f4c5a74b5594cc23fa8fe3b62010ab5" + integrity sha512-IqMCPbihDpbHV4bNws015hU0svIBGyzPjJearwXMGJyungWdblbBcboNojTz9bWOrrJD3zIwmcr5w5c+NH+2+A== + node-addon-api@^3.0.0: version "3.2.1" resolved "https://registry.npmjs.org/node-addon-api/-/node-addon-api-3.2.1.tgz#81325e0a2117789c0128dab65e7e38f07ceba161"