From 5e06a9f45b65f26b971ae7101dc2c2f709a4bba5 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Thu, 21 Jan 2021 15:53:01 +0100 Subject: [PATCH] Refactor into broker --- plugins/scaffolder-backend/package.json | 2 + .../src/scaffolder/tasks/Database.ts | 74 ++++++++++-- .../src/scaffolder/tasks/TaskBroker.test.ts | 63 ++++++++++ .../src/scaffolder/tasks/TaskBroker.ts | 109 ++++++++++++++++++ .../src/scaffolder/tasks/TaskObserver.ts | 15 --- .../src/scaffolder/tasks/taskWorker.ts | 34 ------ .../src/scaffolder/tasks/tasksDispatcher.ts | 29 ----- .../src/scaffolder/tasks/types.ts | 40 +++++-- 8 files changed, 272 insertions(+), 94 deletions(-) create mode 100644 plugins/scaffolder-backend/src/scaffolder/tasks/TaskBroker.test.ts create mode 100644 plugins/scaffolder-backend/src/scaffolder/tasks/TaskBroker.ts delete mode 100644 plugins/scaffolder-backend/src/scaffolder/tasks/TaskObserver.ts delete mode 100644 plugins/scaffolder-backend/src/scaffolder/tasks/taskWorker.ts delete mode 100644 plugins/scaffolder-backend/src/scaffolder/tasks/tasksDispatcher.ts diff --git a/plugins/scaffolder-backend/package.json b/plugins/scaffolder-backend/package.json index 7af9f567e3..97ab723e81 100644 --- a/plugins/scaffolder-backend/package.json +++ b/plugins/scaffolder-backend/package.json @@ -53,7 +53,9 @@ "helmet": "^4.0.0", "isomorphic-git": "^1.8.0", "jsonschema": "^1.2.6", + "luxon": "^1.25.0", "morgan": "^1.10.0", + "p-queue": "^6.3.0", "uuid": "^8.2.0", "winston": "^3.2.1", "yaml": "^1.10.0" diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/Database.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/Database.ts index f446f1e90a..7c3f343b0f 100644 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/Database.ts +++ b/plugins/scaffolder-backend/src/scaffolder/tasks/Database.ts @@ -13,23 +13,79 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -import { Task, Status } from './types'; +import { DbTaskRow, TaskSpec } from './types'; +import { v4 as uuid } from 'uuid'; +import { DateTime } from 'luxon'; -export class Database { - private readonly store = new Map(); +export interface Database { + get(taskId: string): Promise; + // updateTask(task: Task): Promise; + createTask(task: TaskSpec): Promise; + claimTask(): Promise; + heartBeat(runId: string): Promise; +} - get(taskId: string) { - return this.store.get(taskId); +export class InMemoryDatabase implements Database { + private readonly store = new Map(); + + async heartBeat(runId: string): Promise { + let task: DbTaskRow | undefined; + + for (const t of this.store.values()) { + if (t.runId === runId) { + task = t; + } + } + + if (!task) { + throw new Error('No task with matching runId found'); + } + + this.store.set(task.taskId, { + ...task, + lastHeartbeat: DateTime.local().toString(), + }); } - write(task: Task) { - return task; + async claimTask(): Promise { + for (const t of this.store.values()) { + if (t.status === 'OPEN') { + const task: DbTaskRow = { + ...t, + status: 'PROCESSING', + runId: uuid(), + }; + this.store.set(t.taskId, task); + return task; + } + } + throw new Error('No task found'); } - writeStatus(taskId: string, status: Status) { + async createTask(spec: TaskSpec): Promise { + return { + taskId: uuid(), + spec, + status: 'OPEN', + retryCount: 0, + createdAt: new Date().toISOString(), + }; + } + + async get(taskId: string): Promise { const task = this.store.get(taskId); if (task) { - this.store.set(taskId, { ...task, status }); + return task; } + throw new Error(`could not found task ${taskId}`); } + + // async updateTask(task: Task): Promise { + // if (!task.taskId) { + // throw new Error('Task must contain id'); + // } + + // this.store.set(task.taskId, task); + // return task; + // } } diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/TaskBroker.test.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/TaskBroker.test.ts new file mode 100644 index 0000000000..4fb7439bf8 --- /dev/null +++ b/plugins/scaffolder-backend/src/scaffolder/tasks/TaskBroker.test.ts @@ -0,0 +1,63 @@ +/* + * Copyright 2021 Spotify AB + * + * 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 { MemoryTaskBroker, TaskAgent } from './TaskBroker'; + +describe('MemoryTaskBroker', () => { + it('should claim a dispatched work item', async () => { + const broker = new MemoryTaskBroker(); + + await broker.dispatch({}); + await expect(broker.claim()).resolves.toEqual(expect.any(TaskAgent)); + }); + + it('should wait for a dispatched work item', async () => { + const broker = new MemoryTaskBroker(); + + const promise = broker.claim(); + + await expect(Promise.race([promise, 'waiting'])).resolves.toBe('waiting'); + + await broker.dispatch({}); + await expect(promise).resolves.toEqual(expect.any(TaskAgent)); + }); + + it('should dispatch multiple items and claim them in order', async () => { + const broker = new MemoryTaskBroker(); + + await broker.dispatch({ name: 'a' }); + await broker.dispatch({ name: 'b' }); + await broker.dispatch({ name: 'c' }); + + const taskA = await broker.claim(); + const taskB = await broker.claim(); + const taskC = await broker.claim(); + await expect(taskA).toEqual(expect.any(TaskAgent)); + await expect(taskB).toEqual(expect.any(TaskAgent)); + await expect(taskC).toEqual(expect.any(TaskAgent)); + await expect(taskA.spec.name).toBe('a'); + await expect(taskB.spec.name).toBe('b'); + await expect(taskC.spec.name).toBe('c'); + }); + + it('should complete a task', async () => { + const broker = new MemoryTaskBroker(); + + await broker.dispatch({}); + const task = await broker.claim(); + await task.complete('COMPLETED'); + }); +}); diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/TaskBroker.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/TaskBroker.ts new file mode 100644 index 0000000000..322e669373 --- /dev/null +++ b/plugins/scaffolder-backend/src/scaffolder/tasks/TaskBroker.ts @@ -0,0 +1,109 @@ +/* + * Copyright 2021 Spotify AB + * + * 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 { + CompletedTaskState, + Task, + TaskSpec, + TaskBroker, + Status, +} from './types'; +import { v4 as uuid } from 'uuid'; +import { InMemoryDatabase } from './database'; + +export class TaskAgent implements Task { + private heartbeartInterval?: ReturnType; + + static create(state: TaskState, db: InMemoryDatabase) { + const agent = new TaskAgent(state, db); + agent.start(); + return agent; + } + + // Runs heartbeat internally + private constructor( + private readonly state: TaskState, + private readonly db: InMemoryDatabase, + ) {} + + get spec() { + return this.state.spec; + } + + async emitLog(message: string): Promise { + throw new Error('Method not implemented.'); + } + + async complete(result: CompletedTaskState): Promise { + this.state.status = result === 'FAILED' ? 'COMPLETED' : 'FAILED'; + } + + private start() { + this.heartbeartInterval = setInterval(() => { + const runId = 'iiiid'; + this.db.heartBeat(runId); + }, 4269); + } +} + +interface TaskState { + spec: TaskSpec; + status: Status; + runId: string | undefined; +} + +function defer() { + let resolve = () => {}; + const promise = new Promise(_resolve => { + resolve = _resolve; + }); + return { promise, resolve }; +} + +export class MemoryTaskBroker implements TaskBroker { + private readonly db = new InMemoryDatabase(); + private readonly tasks = new Array(); + private deferredDispatch = defer(); + + async claim(): Promise { + for (;;) { + const pendingTask = await this.db.claimTask(); + if (pendingTask) { + return TaskAgent.create(pendingTask, this.db); + } + + await this.waitForDispatch(); + } + } + + async dispatch(spec: TaskSpec): Promise { + this.tasks.push({ + spec, + status: 'OPEN', + runId: undefined, + }); + this.signalDispatch(); + } + + private waitForDispatch() { + return this.deferredDispatch.promise; + } + + private signalDispatch() { + this.deferredDispatch.resolve(); + this.deferredDispatch = defer(); + } +} diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/TaskObserver.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/TaskObserver.ts deleted file mode 100644 index 863d6e76e1..0000000000 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/TaskObserver.ts +++ /dev/null @@ -1,15 +0,0 @@ -/* - * Copyright 2021 Spotify AB - * - * 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. - */ diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/taskWorker.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/taskWorker.ts deleted file mode 100644 index 4b1754523c..0000000000 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/taskWorker.ts +++ /dev/null @@ -1,34 +0,0 @@ -/* - * Copyright 2021 Spotify AB - * - * 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 { Status, ClaimResponse } from './types'; -import { Database } from './Database'; - -export class TaskWorker { - static async fromConfig() {} - - constructor(private readonly database: Database) {} - - claim(): Promise { - return Promise.resolve(undefined); - } - - setStatus(taskId: string, status: Status) { - this.database.writeStatus(taskId, status); - } - - heartbeat(runId: number) {} -} diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/tasksDispatcher.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/tasksDispatcher.ts deleted file mode 100644 index c3dfaa3c88..0000000000 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/tasksDispatcher.ts +++ /dev/null @@ -1,29 +0,0 @@ -/* - * Copyright 2021 Spotify AB - * - * 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 { Task } from './types'; -import { Database } from './Database'; - -export class taskDispatcher { - static async fromConfig(config: { database: Database }) {} - - constructor(private readonly db: Database) {} - - dispatch(task: Task): Promise { - this.db.write(task); - return Promise.resolve('uuid'); - } -} diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/types.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/types.ts index 000e292e16..bd8bb0ade5 100644 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/types.ts +++ b/plugins/scaffolder-backend/src/scaffolder/tasks/types.ts @@ -14,17 +14,43 @@ * limitations under the License. */ -export type Status = 'OPEN' | 'PROCESSING' | 'FAILED' | 'CANCELLED'; -export type Task = { +export type Status = + | 'OPEN' + | 'PROCESSING' + | 'FAILED' + | 'CANCELLED' + | 'COMPLETED'; + +export type CompletedTaskState = 'FAILED' | 'COMPLETED'; + +export type DbTaskRow = { taskId: string; - metadata: string; + spec: TaskSpec; status: Status; - lastHeartbeat: string; + lastHeartbeat?: string; retryCount: number; createdAt: string; + runId?: string; +}; + +export type DbTaskEventRow = { + id: number; + runId?: number; + stageName: string; + createdAt: string; }; -export type ClaimResponse = { - runId: number; - task: Task; +export type TaskSpec = { + metadata: string; }; + +export interface Task { + spec: TaskSpec; + emitLog(message: string): Promise; + complete(result: CompletedTaskState): Promise; +} + +export interface TaskBroker { + claim(): Promise; + dispatch(spec: TaskSpec): Promise; +}