diff --git a/plugins/scaffolder-backend/package.json b/plugins/scaffolder-backend/package.json index 5d024c2798..bdecf6196e 100644 --- a/plugins/scaffolder-backend/package.json +++ b/plugins/scaffolder-backend/package.json @@ -58,7 +58,8 @@ "p-queue": "^6.3.0", "uuid": "^8.2.0", "winston": "^3.2.1", - "yaml": "^1.10.0" + "yaml": "^1.10.0", + "luxon": "^1.25.0" }, "devDependencies": { "@backstage/cli": "^0.5.0", diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/DatabaseTaskStore.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/DatabaseTaskStore.ts index e6de1735d8..1d99cd2958 100644 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/DatabaseTaskStore.ts +++ b/plugins/scaffolder-backend/src/scaffolder/tasks/DatabaseTaskStore.ts @@ -20,8 +20,7 @@ import { NotFoundError, resolvePackagePath, } from '@backstage/backend-common'; -import Knex, { Transaction } from 'knex'; -import { Logger } from 'winston'; +import Knex from 'knex'; import { v4 as uuid } from 'uuid'; import { DbTaskEventRow, @@ -121,6 +120,7 @@ export class DatabaseTaskStore implements TaskStore { .update({ status: 'processing', run_id: runId, + last_heartbeat_at: this.db.fn.now(), }); if (updateCount < 1) { @@ -155,6 +155,18 @@ export class DatabaseTaskStore implements TaskStore { } } + async listStaleTasks(): Promise<{ tasks: DbTaskRow }> { + const rows = await this.db('tasks') + .where('status', 'processing') + .andWhere( + 'last_heartbeat_at', + '<', + this.db.type === 'sqlite' + ? this.db.raw("datetime('now', '-2 seconds')") + : this.db.raw("dateadd('second', -2, now())"), + ); + } + async setStatus(runId: string, status: Status): Promise { let oldStatus: string; if (status === 'failed' || status === 'completed') { @@ -216,16 +228,17 @@ export class DatabaseTaskStore implements TaskStore { taskId, after, }: TaskStoreGetEventsOptions): Promise<{ events: DbTaskEventRow[] }> { - let query = this.db('task_events').where({ - task_id: taskId, - }); - if (typeof after === 'number') { - query = query - .where('task_events.id', '>', after) - .orWhere({ event_type: 'completion' }); - } + const rawEvents = await this.db('task_events') + .where({ + task_id: taskId, + }) + .andWhere(builder => { + if (typeof after === 'number') { + builder.where('id', '>', after).orWhere('event_type', 'completion'); + } + }) + .select(); - const rawEvents = await query.select(); const events = rawEvents.map(event => { try { const body = JSON.parse(event.body) as JsonObject; diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.test.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.test.ts index 5bbe91ea87..ebbac1db5f 100644 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.test.ts +++ b/plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.test.ts @@ -14,47 +14,54 @@ * limitations under the License. */ -import { TemplateEntityV1alpha1 } from '@backstage/catalog-model'; -import { TemplaterValues } from '../stages'; -import { MemoryTaskStore } from './MemoryTaskStore'; +import { SingleConnectionDatabaseManager } from '@backstage/backend-common'; +import { ConfigReader } from '@backstage/config'; +import { DatabaseTaskStore } from './DatabaseTaskStore'; import { StorageTaskBroker, TaskAgent } from './StorageTaskBroker'; +import { TaskStore, TaskSpec, DbTaskEventRow } from './types'; + +async function createStore(): Promise { + const manager = SingleConnectionDatabaseManager.fromConfig( + new ConfigReader({ + backend: { + database: { + client: 'sqlite3', + connection: ':memory:', + }, + }, + }), + ).forPlugin('scaffolder'); + return await DatabaseTaskStore.create(await manager.getClient()); +} describe('StorageTaskBroker', () => { - const storage = new MemoryTaskStore(); - const broker = new StorageTaskBroker(storage); + let storage: TaskStore; - const taskSpec = { - values: {} as TemplaterValues, - template: {} as TemplateEntityV1alpha1, - }; + beforeAll(async () => { + storage = await createStore(); + }); it('should claim a dispatched work item', async () => { - await broker.dispatch(taskSpec); + const broker = new StorageTaskBroker(storage); + await broker.dispatch({ steps: [] }); await expect(broker.claim()).resolves.toEqual(expect.any(TaskAgent)); }); it('should wait for a dispatched work item', async () => { + const broker = new StorageTaskBroker(storage); const promise = broker.claim(); await expect(Promise.race([promise, 'waiting'])).resolves.toBe('waiting'); - await broker.dispatch(taskSpec); + await broker.dispatch({ steps: [] }); await expect(promise).resolves.toEqual(expect.any(TaskAgent)); }); it('should dispatch multiple items and claim them in order', async () => { - await broker.dispatch({ - values: { owner: 'a' } as TemplaterValues, - template: {} as TemplateEntityV1alpha1, - }); - await broker.dispatch({ - values: { owner: 'b' } as TemplaterValues, - template: {} as TemplateEntityV1alpha1, - }); - await broker.dispatch({ - values: { owner: 'c' } as TemplaterValues, - template: {} as TemplateEntityV1alpha1, - }); + const broker = new StorageTaskBroker(storage); + await broker.dispatch({ steps: [{ id: 'a' }] } as TaskSpec); + await broker.dispatch({ steps: [{ id: 'b' }] } as TaskSpec); + await broker.dispatch({ steps: [{ id: 'c' }] } as TaskSpec); const taskA = await broker.claim(); const taskB = await broker.claim(); @@ -62,13 +69,14 @@ describe('StorageTaskBroker', () => { 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.values.owner).toBe('a'); - await expect(taskB.spec.values.owner).toBe('b'); - await expect(taskC.spec.values.owner).toBe('c'); + await expect(taskA.spec.steps[0].id).toBe('a'); + await expect(taskB.spec.steps[0].id).toBe('b'); + await expect(taskC.spec.steps[0].id).toBe('c'); }); it('should complete a task', async () => { - const dispatchResult = await broker.dispatch(taskSpec); + const broker = new StorageTaskBroker(storage); + const dispatchResult = await broker.dispatch({ steps: [] }); const task = await broker.claim(); await task.complete('completed'); const taskRow = await storage.get(dispatchResult.taskId); @@ -76,10 +84,91 @@ describe('StorageTaskBroker', () => { }); it('should fail a task', async () => { - const dispatchResult = await broker.dispatch(taskSpec); + const broker = new StorageTaskBroker(storage); + const dispatchResult = await broker.dispatch({ steps: [] }); const task = await broker.claim(); await task.complete('failed'); const taskRow = await storage.get(dispatchResult.taskId); expect(taskRow.status).toBe('failed'); }); + + it('multiple brokers should be able to observe a single task', async () => { + const broker1 = new StorageTaskBroker(storage); + const broker2 = new StorageTaskBroker(storage); + + const { taskId } = await broker1.dispatch({ steps: [] }); + + const logPromise = new Promise(resolve => { + const observedEvents = new Array(); + + broker2.observe({ taskId, after: undefined }, (_err, { events }) => { + observedEvents.push(...events); + if (events.some(e => e.type === 'completion')) { + resolve(observedEvents); + } + }); + }); + const task = await broker1.claim(); + await task.emitLog('log 1'); + await task.emitLog('log 2'); + await task.emitLog('log 3'); + await task.complete('completed'); + + const logs = await logPromise; + expect(logs.map(l => l.body.message)).toEqual([ + 'log 1', + 'log 2', + 'log 3', + 'Run completed with status: completed', + ]); + + const afterLogs = await new Promise(resolve => { + broker2.observe({ taskId, after: logs[1].id }, (_err, { events }) => + resolve(events.map(e => e.body.message as string)), + ); + }); + expect(afterLogs).toEqual([ + 'log 3', + 'Run completed with status: completed', + ]); + }); + + it('should heartbeat', async () => { + const broker = new StorageTaskBroker(storage); + const { taskId } = await broker.dispatch({ steps: [] }); + const task = await broker.claim(); + + const initialTask = await storage.get(taskId); + + for (;;) { + const maybeTask = await storage.get(taskId); + if (maybeTask.lastHeartbeat !== initialTask.lastHeartbeat) { + break; + } + await new Promise(resolve => setTimeout(resolve, 50)); + } + await task.complete('completed'); + expect.assertions(0); + }); + + it('should be cancelled if heartbeat stops', async () => { + const broker = new StorageTaskBroker(storage); + const { taskId } = await broker.dispatch({ steps: [] }); + console.log('DEBUG: taskId =', taskId); + const task = await broker.claim(); + clearInterval((task as any).heartbeatInterval); + + setTimeout(() => { + storage.listStaleTasks(); + }, 4000); + + for (;;) { + const maybeTask = await storage.get(taskId); + if (maybeTask.status === 'cancelled') { + break; + } + await new Promise(resolve => setTimeout(resolve, 50)); + } + expect.assertions(0); + }); }); diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.ts index 6eedc023f8..8c0d63ee71 100644 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.ts +++ b/plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.ts @@ -25,7 +25,7 @@ import { } from './types'; export class TaskAgent implements Task { - private heartbeartInterval?: ReturnType; + private heartbeatInterval?: ReturnType; static create(state: TaskState, storage: TaskStore) { const agent = new TaskAgent(state, storage); @@ -67,13 +67,13 @@ export class TaskAgent implements Task { body: { message: `Run completed with status: ${result}` }, type: 'completion', }); - if (this.heartbeartInterval) { - clearInterval(this.heartbeartInterval); + if (this.heartbeatInterval) { + clearInterval(this.heartbeatInterval); } } private start() { - this.heartbeartInterval = setInterval(() => { + this.heartbeatInterval = setInterval(() => { if (!this.state.runId) { throw new Error('no run id provided'); } @@ -131,7 +131,10 @@ export class StorageTaskBroker implements TaskBroker { taskId: string; after: number | undefined; }, - callback: (result: { events: DbTaskEventRow[] }) => void, + callback: ( + error: Error | undefined, + result: { events: DbTaskEventRow[] }, + ) => void, ): () => void { const { taskId } = options; @@ -148,9 +151,9 @@ export class StorageTaskBroker implements TaskBroker { if (events.length) { after = events[events.length - 1].id; try { - callback(result); + callback(undefined, result); } catch (error) { - console.log('DEBUG: error =', error); + callback(error, { events: [] }); } } diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/types.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/types.ts index f9081af6b6..dd9a1ad9ad 100644 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/types.ts +++ b/plugins/scaffolder-backend/src/scaffolder/tasks/types.ts @@ -86,6 +86,7 @@ export interface TaskStore { createTask(task: TaskSpec): Promise<{ taskId: string }>; claimTask(): Promise; heartbeat(runId: string): Promise; + listStaleTasks(): Promise<{ tasks: DbTaskRow }>; setStatus(runId: string, status: Status): Promise; emit({ taskId, runId, body, type }: TaskStoreEmitOptions): Promise; getEvents({ diff --git a/plugins/scaffolder-backend/src/service/router.test.ts b/plugins/scaffolder-backend/src/service/router.test.ts index 31f62febd2..8a0fa6cba6 100644 --- a/plugins/scaffolder-backend/src/service/router.test.ts +++ b/plugins/scaffolder-backend/src/service/router.test.ts @@ -16,6 +16,7 @@ const mockAccess = jest.fn(); jest.doMock('fs-extra', () => ({ + access: mockAccess, promises: { access: mockAccess, }, @@ -27,7 +28,11 @@ jest.doMock('fs-extra', () => ({ remove: jest.fn(), })); -import { getVoidLogger } from '@backstage/backend-common'; +import { + SingleConnectionDatabaseManager, + PluginDatabaseManager, + getVoidLogger, +} from '@backstage/backend-common'; import { ConfigReader } from '@backstage/config'; import express from 'express'; import request from 'supertest'; @@ -41,6 +46,19 @@ const generateEntityClient: any = (template: any) => ({ findTemplate: () => Promise.resolve(template), }); +function createDatabase(): PluginDatabaseManager { + return SingleConnectionDatabaseManager.fromConfig( + new ConfigReader({ + backend: { + database: { + client: 'sqlite3', + connection: ':memory:', + }, + }, + }), + ).forPlugin('scaffolder'); +} + describe('createRouter - working directory', () => { const mockPrepare = jest.fn(); const mockPreparers = new Preparers(); @@ -78,7 +96,6 @@ describe('createRouter - working directory', () => { }; const mockedEntityClient = generateEntityClient(template); - it('should throw an error when working directory does not exist or is not writable', async () => { mockAccess.mockImplementation(() => { throw new Error('access error'); @@ -93,6 +110,7 @@ describe('createRouter - working directory', () => { config: new ConfigReader(workDirConfig('/path')), dockerClient: new Docker(), entityClient: mockedEntityClient, + database: createDatabase(), }), ).rejects.toThrow('access error'); }); @@ -106,6 +124,7 @@ describe('createRouter - working directory', () => { config: new ConfigReader(workDirConfig('/path')), dockerClient: new Docker(), entityClient: mockedEntityClient, + database: createDatabase(), }); const app = express().use(router); @@ -134,6 +153,7 @@ describe('createRouter - working directory', () => { config: new ConfigReader({}), dockerClient: new Docker(), entityClient: mockedEntityClient, + database: createDatabase(), }); const app = express().use(router); @@ -203,6 +223,7 @@ describe('createRouter', () => { config: new ConfigReader({}), dockerClient: new Docker(), entityClient: generateEntityClient(template), + database: createDatabase(), }); app = express().use(router); }); diff --git a/plugins/scaffolder-backend/src/service/router.ts b/plugins/scaffolder-backend/src/service/router.ts index 23af538df8..6eb112fa3c 100644 --- a/plugins/scaffolder-backend/src/service/router.ts +++ b/plugins/scaffolder-backend/src/service/router.ts @@ -164,7 +164,11 @@ export async function createRouter( // After client opens connection send all nests as string const unsubscribe = taskBroker.observe( { taskId, after }, - ({ events }) => { + (error, { events }) => { + logger.error( + `Received error from event stream when observing task ${taskId}`, + error, + ); for (const event of events) { res.write(`event:${JSON.stringify(event)}\n\n`); if (event.type === 'completion') {