Implement the locks part of the task manager

Signed-off-by: Fredrik Adelöw <freben@gmail.com>

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',
+  },
+};

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');
+};

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"

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,
     };

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<void> {
+  await knex.migrate.latest({
+    directory: migrationsDir,
+    tableName: 'knex_migrations_backstage_backend_common',
+  });
+}

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;
+};

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';

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<DatabaseManager> {
+    const knex = await databases.init(databaseId);
+    const databaseManager: Partial<DatabaseManager> = {
+      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();
+      },
+    );
+  });
+});

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<Knex>,
+  ) {}
+
+  async acquireLock(
+    idWithoutPrefix: string,
+    options: {
+      timeout: Duration;
+    },
+  ): Promise<
+    | { acquired: false }
+    | { acquired: true; release: () => void | Promise<void> }
+  > {
+    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<DbTaskLocksRow>('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<DbTaskLocksRow>(
+      '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<DbTaskLocksRow>('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<void>,
+  ): Promise<{ release: () => Promise<void> }> {
+    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;
+      }),
+    );
+  }
+}

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';

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<void> }
+  >;
+
+  scheduleTask(
+    id: string,
+    options: {
+      timeout: Duration;
+      frequency: Duration;
+      initialDelay?: Duration;
+    },
+    fn: () => Promise<void>,
+  ): Promise<{ release: () => Promise<void> }>;
+}

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`,
+    );
+  }
+}

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'));

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;

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"