diff --git a/packages/backend/src/plugins/scaffolder.ts b/packages/backend/src/plugins/scaffolder.ts index 4e2257a46c..723165ff82 100644 --- a/packages/backend/src/plugins/scaffolder.ts +++ b/packages/backend/src/plugins/scaffolder.ts @@ -30,6 +30,7 @@ import Docker from 'dockerode'; export default async function createPlugin({ logger, config, + database, }: PluginEnvironment) { const cookiecutterTemplater = new CookieCutter(); const craTemplater = new CreateReactAppTemplater(); @@ -54,5 +55,6 @@ export default async function createPlugin({ config, dockerClient, entityClient, + database, }); } diff --git a/packages/create-app/templates/default-app/packages/backend/src/plugins/scaffolder.ts b/packages/create-app/templates/default-app/packages/backend/src/plugins/scaffolder.ts index c8bd3e5012..d68f90ce08 100644 --- a/packages/create-app/templates/default-app/packages/backend/src/plugins/scaffolder.ts +++ b/packages/create-app/templates/default-app/packages/backend/src/plugins/scaffolder.ts @@ -14,6 +14,7 @@ import Docker from 'dockerode'; export default async function createPlugin({ logger, config, + database, }: PluginEnvironment) { const cookiecutterTemplater = new CookieCutter(); const craTemplater = new CreateReactAppTemplater(); @@ -38,5 +39,6 @@ export default async function createPlugin({ config, dockerClient, entityClient, + database, }); } diff --git a/plugins/scaffolder-backend/migrations/20210120143715_init.js b/plugins/scaffolder-backend/migrations/20210120143715_init.js index d2535bfa86..bb65e1e2fb 100644 --- a/plugins/scaffolder-backend/migrations/20210120143715_init.js +++ b/plugins/scaffolder-backend/migrations/20210120143715_init.js @@ -23,10 +23,6 @@ exports.up = async function up(knex) { await knex.schema.createTable('tasks', table => { table.comment('The table of scaffolder tasks'); table.uuid('id').primary().notNullable().comment('The ID of the task'); - table - .text('context') - .notNullable() - .comment('A JSON object with task specific context information'); table .text('spec') .notNullable() @@ -41,6 +37,7 @@ exports.up = async function up(knex) { .comment('The current run ID of the task'); table .dateTime('created_at') + .defaultTo(knex.fn.now()) .notNullable() .comment('The timestamp when this task was created'); table @@ -53,6 +50,7 @@ exports.up = async function up(knex) { .defaultTo(0) .comment('The number of times that this task has been attempted'); }); + await knex.schema.createTable('task_events', table => { table.comment('The event stream a given task'); table @@ -72,16 +70,16 @@ exports.up = async function up(knex) { .nullable() .comment('The run ID of the task that this event applies to'); table - .text('stage_name') - .nullable() - .comment('The stage of the task that this event applies to'); + .text('body') + .notNullable() + .comment('The JSON encoded body of the event'); table.text('event_type').notNullable().comment('The type of event'); table .dateTime('created_at') .defaultTo(knex.fn.now()) .notNullable() .comment('The timestamp when this event was generated'); - table.text('text').notNullable().comment('The text of the event'); + table.index(['task_id'], 'task_events_task_id_idx'); }); }; diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/DatabaseTaskStore.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/DatabaseTaskStore.ts new file mode 100644 index 0000000000..e6de1735d8 --- /dev/null +++ b/plugins/scaffolder-backend/src/scaffolder/tasks/DatabaseTaskStore.ts @@ -0,0 +1,248 @@ +/* + * 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 { JsonObject } from '@backstage/config'; +import { + ConflictError, + NotFoundError, + resolvePackagePath, +} from '@backstage/backend-common'; +import Knex, { Transaction } from 'knex'; +import { Logger } from 'winston'; +import { v4 as uuid } from 'uuid'; +import { + DbTaskEventRow, + DbTaskRow, + Status, + TaskEventType, + TaskSpec, + TaskStore, + TaskStoreEmitOptions, + TaskStoreGetEventsOptions, +} from './types'; + +const migrationsDir = resolvePackagePath( + '@backstage/plugin-scaffolder-backend', + 'migrations', +); + +export type RawDbTaskRow = { + id: string; + spec: string; + status: Status; + last_heartbeat_at?: string; + retry_count: number; + created_at: string; + run_id?: string; +}; + +export type RawDbTaskEventRow = { + id: number; + run_id: string; + task_id: string; + body: string; + event_type: TaskEventType; + created_at: string; +}; + +export class DatabaseTaskStore implements TaskStore { + static async create(knex: Knex): Promise { + await knex.migrate.latest({ + directory: migrationsDir, + }); + return new DatabaseTaskStore(knex); + } + + constructor(private readonly db: Knex) {} + + async get(taskId: string): Promise { + const [result] = await this.db('tasks') + .where({ id: taskId }) + .select(); + if (!result) { + throw new NotFoundError(`No task with id '${taskId}' found`); + } + try { + const spec = JSON.parse(result.spec); + return { + id: result.id, + spec, + status: result.status, + lastHeartbeat: result.last_heartbeat_at, + retryCount: result.retry_count, + createdAt: result.created_at, + runId: result.run_id, + }; + } catch (error) { + throw new Error(`Failed to parse spec of task '${taskId}', ${error}`); + } + } + + async createTask(spec: TaskSpec): Promise<{ taskId: string }> { + const taskId = uuid(); + await this.db('tasks').insert({ + id: taskId, + spec: JSON.stringify(spec), + status: 'open', + retry_count: 0, + }); + return { taskId }; + } + + async claimTask(): Promise { + return this.db.transaction(async tx => { + const [task] = await tx('tasks') + .where({ + status: 'open', + }) + .limit(1) + .select(); + + if (!task) { + return undefined; + } + + const runId = uuid(); + const updateCount = await tx('tasks') + .where({ id: task.id, status: 'open' }) + .update({ + status: 'processing', + run_id: runId, + }); + + if (updateCount < 1) { + return undefined; + } + + try { + const spec = JSON.parse(task.spec); + return { + id: task.id, + spec, + status: 'processing', + lastHeartbeat: task.last_heartbeat_at, + retryCount: task.retry_count, + createdAt: task.created_at, + runId: runId, + }; + } catch (error) { + throw new Error(`Failed to parse spec of task '${task.id}', ${error}`); + } + }); + } + + async heartbeat(runId: string): Promise { + const updateCount = await this.db('tasks') + .where({ run_id: runId, status: 'processing' }) + .update({ + last_heartbeat_at: this.db.fn.now(), + }); + if (updateCount === 0) { + throw new Error(`No running task with runId ${runId} found`); + } + } + + async setStatus(runId: string, status: Status): Promise { + let oldStatus: string; + if (status === 'failed' || status === 'completed') { + oldStatus = 'processing'; + } else { + throw new Error( + `Invalid status update of run '${runId}' to status '${status}'`, + ); + } + await this.db.transaction(async tx => { + const [task] = await tx('tasks') + .where({ + run_id: runId, + }) + .limit(1) + .select(); + + if (!task) { + throw new Error(`No task with runId ${runId} found`); + } + if (task.status !== oldStatus) { + throw new ConflictError( + `Refusing to update status of run '${runId}' to status '${status}' ` + + `as it is currently '${task.status}', expected '${oldStatus}'`, + ); + } + const updateCount = await tx('tasks') + .where({ + run_id: runId, + status: oldStatus, + }) + .update({ + status, + }); + if (updateCount !== 1) { + throw new Error( + `Failed to update status to '${status}' for runId ${runId}`, + ); + } + }); + } + + async emit({ + taskId, + runId, + body, + type, + }: TaskStoreEmitOptions): Promise { + const serliazedBody = JSON.stringify(body); + await this.db('task_events').insert({ + task_id: taskId, + run_id: runId, + event_type: type, + body: serliazedBody, + }); + } + + async getEvents({ + 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 query.select(); + const events = rawEvents.map(event => { + try { + const body = JSON.parse(event.body) as JsonObject; + return { + id: event.id, + runId: event.run_id, + taskId: event.task_id, + body, + type: event.event_type, + createdAt: event.created_at, + }; + } catch (error) { + throw new Error( + `Failed to parse event body from event taskId=${taskId} id=${event.id}, ${error}`, + ); + } + }); + return { events }; + } +} diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/MemoryDatabase.test.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/MemoryTaskStore.test.ts similarity index 100% rename from plugins/scaffolder-backend/src/scaffolder/tasks/MemoryDatabase.test.ts rename to plugins/scaffolder-backend/src/scaffolder/tasks/MemoryTaskStore.test.ts diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/MemoryDatabase.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/MemoryTaskStore.ts similarity index 73% rename from plugins/scaffolder-backend/src/scaffolder/tasks/MemoryDatabase.ts rename to plugins/scaffolder-backend/src/scaffolder/tasks/MemoryTaskStore.ts index b2d1cf667b..e181edf524 100644 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/MemoryDatabase.ts +++ b/plugins/scaffolder-backend/src/scaffolder/tasks/MemoryTaskStore.ts @@ -19,35 +19,17 @@ import { DbTaskEventRow, Status, TaskSpec, - TaskEventType, + TaskStore, + TaskStoreGetEventsOptions, + TaskStoreEmitOptions, } from './types'; import { v4 as uuid } from 'uuid'; -export interface Database { - get(taskId: string): Promise; - createTask(task: TaskSpec): Promise; - claimTask(): Promise; - heartbeat(runId: string): Promise; - setStatus(taskId: string, status: Status): Promise; -} - -type EmitOptions = { - taskId: string; - runId: string; - body: string; - type: TaskEventType; -}; - -type ReadOptions = { - taskId: string; - after?: number | undefined; -}; - -export class MemoryDatabase implements Database { +export class MemoryTaskStore implements TaskStore { private readonly store = new Map(); private readonly events = new Array(); - async emit({ taskId, runId, body, type }: EmitOptions) { + async emit({ taskId, runId, body, type }: TaskStoreEmitOptions) { this.events.push({ id: this.events.length, taskId, @@ -61,7 +43,7 @@ export class MemoryDatabase implements Database { async getEvents({ taskId, after, - }: ReadOptions): Promise<{ events: DbTaskEventRow[] }> { + }: TaskStoreGetEventsOptions): Promise<{ events: DbTaskEventRow[] }> { const events = this.events.filter(event => { if (event.taskId !== taskId) { return false; @@ -89,7 +71,7 @@ export class MemoryDatabase implements Database { throw new Error('No task with matching runId found'); } - this.store.set(task.taskId, { + this.store.set(task.id, { ...task, lastHeartbeat: new Date().toISOString(), }); @@ -103,23 +85,23 @@ export class MemoryDatabase implements Database { status: 'processing', runId: uuid(), }; - this.store.set(t.taskId, task); + this.store.set(t.id, task); return task; } } return undefined; } - async createTask(spec: TaskSpec): Promise { + async createTask(spec: TaskSpec): Promise<{ taskId: string }> { const taskRow = { - taskId: uuid(), + id: uuid(), spec, status: 'open' as Status, retryCount: 0, createdAt: new Date().toISOString(), }; - this.store.set(taskRow.taskId, taskRow); - return taskRow; + this.store.set(taskRow.id, taskRow); + return { taskId: taskRow.id }; } async get(taskId: string): Promise { @@ -135,6 +117,6 @@ export class MemoryDatabase implements Database { if (!task) { throw new Error(`no task found`); } - this.store.set(task.taskId, { ...task, status }); + this.store.set(task.id, { ...task, status }); } } diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/MemoryTaskBroker.test.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.test.ts similarity index 90% rename from plugins/scaffolder-backend/src/scaffolder/tasks/MemoryTaskBroker.test.ts rename to plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.test.ts index 686a6d9132..5bbe91ea87 100644 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/MemoryTaskBroker.test.ts +++ b/plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.test.ts @@ -15,13 +15,13 @@ */ import { TemplateEntityV1alpha1 } from '@backstage/catalog-model'; -import { TemplaterValues } from './actions/templater/types'; -import { MemoryDatabase } from './MemoryDatabase'; -import { MemoryTaskBroker, TaskAgent } from './MemoryTaskBroker'; +import { TemplaterValues } from '../stages'; +import { MemoryTaskStore } from './MemoryTaskStore'; +import { StorageTaskBroker, TaskAgent } from './StorageTaskBroker'; -describe('MemoryTaskBroker', () => { - const storage = new MemoryDatabase(); - const broker = new MemoryTaskBroker(storage); +describe('StorageTaskBroker', () => { + const storage = new MemoryTaskStore(); + const broker = new StorageTaskBroker(storage); const taskSpec = { values: {} as TemplaterValues, diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/MemoryTaskBroker.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.ts similarity index 90% rename from plugins/scaffolder-backend/src/scaffolder/tasks/MemoryTaskBroker.ts rename to plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.ts index 6937c535dc..6eedc023f8 100644 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/MemoryTaskBroker.ts +++ b/plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.ts @@ -18,16 +18,16 @@ import { CompletedTaskState, Task, TaskSpec, + TaskStore, TaskBroker, DispatchResult, DbTaskEventRow, } from './types'; -import { MemoryDatabase } from './MemoryDatabase'; export class TaskAgent implements Task { private heartbeartInterval?: ReturnType; - static create(state: TaskState, storage: MemoryDatabase) { + static create(state: TaskState, storage: TaskStore) { const agent = new TaskAgent(state, storage); agent.start(); return agent; @@ -36,7 +36,7 @@ export class TaskAgent implements Task { // Runs heartbeat internally private constructor( private readonly state: TaskState, - private readonly storage: MemoryDatabase, + private readonly storage: TaskStore, ) {} get spec() { @@ -51,20 +51,20 @@ export class TaskAgent implements Task { await this.storage.emit({ taskId: this.state.taskId, runId: this.state.runId, - body: message, + body: { message }, type: 'log', }); } async complete(result: CompletedTaskState): Promise { await this.storage.setStatus( - this.state.taskId, + this.state.runId, result === 'failed' ? 'failed' : 'completed', ); this.storage.emit({ taskId: this.state.taskId, runId: this.state.runId, - body: `Run completed with status: ${result}`, + body: { message: `Run completed with status: ${result}` }, type: 'completion', }); if (this.heartbeartInterval) { @@ -96,8 +96,8 @@ function defer() { return { promise, resolve }; } -export class MemoryTaskBroker implements TaskBroker { - constructor(private readonly storage: MemoryDatabase) {} +export class StorageTaskBroker implements TaskBroker { + constructor(private readonly storage: TaskStore) {} private deferredDispatch = defer(); async claim(): Promise { @@ -107,7 +107,7 @@ export class MemoryTaskBroker implements TaskBroker { return TaskAgent.create( { runId: pendingTask.runId!, - taskId: pendingTask.taskId, + taskId: pendingTask.id, spec: pendingTask.spec, }, this.storage, diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/TaskWorker.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/TaskWorker.ts index c80a1b81bf..64cfdc6f96 100644 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/TaskWorker.ts +++ b/plugins/scaffolder-backend/src/scaffolder/tasks/TaskWorker.ts @@ -73,6 +73,7 @@ export class TaskWorker { // Give us some time to curl observe task.emitLog('Task claimed, waiting ...'); await new Promise(resolve => setTimeout(resolve, 5000)); + console.log('DEBUG: task.spec =', JSON.stringify(task.spec, null, 2)); task.emitLog(`Starting up work with ${task.spec.steps.length} steps`); const outputs: { [name: string]: JsonValue } = {}; @@ -106,7 +107,7 @@ export class TaskWorker { await task.complete('completed'); } catch (error) { - task.emitLog(error); + task.emitLog(String(error.stack)); await task.complete('failed'); } } diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/index.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/index.ts index 1cc9842fed..7b6552a3e0 100644 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/index.ts +++ b/plugins/scaffolder-backend/src/scaffolder/tasks/index.ts @@ -14,6 +14,7 @@ * limitations under the License. */ -export { MemoryDatabase } from './MemoryDatabase'; -export { MemoryTaskBroker } from './MemoryTaskBroker'; +export { MemoryTaskStore } from './MemoryTaskStore'; +export { DatabaseTaskStore } from './DatabaseTaskStore'; +export { StorageTaskBroker } from './StorageTaskBroker'; export { TaskWorker } from './TaskWorker'; diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/types.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/types.ts index d3c08085fd..f9081af6b6 100644 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/types.ts +++ b/plugins/scaffolder-backend/src/scaffolder/tasks/types.ts @@ -14,7 +14,7 @@ * limitations under the License. */ -import { JsonValue } from '@backstage/config'; +import { JsonValue, JsonObject } from '@backstage/config'; export type Status = | 'open' @@ -26,7 +26,7 @@ export type Status = export type CompletedTaskState = 'failed' | 'completed'; export type DbTaskRow = { - taskId: string; + id: string; spec: TaskSpec; status: Status; lastHeartbeat?: string; @@ -40,7 +40,7 @@ export type DbTaskEventRow = { id: number; runId: string; taskId: string; - body: string; + body: JsonObject; type: TaskEventType; createdAt: string; }; @@ -69,3 +69,27 @@ export interface TaskBroker { claim(): Promise; dispatch(spec: TaskSpec): Promise; } + +export type TaskStoreEmitOptions = { + taskId: string; + runId: string; + body: JsonObject; + type: TaskEventType; +}; + +export type TaskStoreGetEventsOptions = { + taskId: string; + after?: number | undefined; +}; +export interface TaskStore { + get(taskId: string): Promise; + createTask(task: TaskSpec): Promise<{ taskId: string }>; + claimTask(): Promise; + heartbeat(runId: string): Promise; + setStatus(runId: string, status: Status): Promise; + emit({ taskId, runId, body, type }: TaskStoreEmitOptions): Promise; + getEvents({ + taskId, + after, + }: TaskStoreGetEventsOptions): Promise<{ events: DbTaskEventRow[] }>; +} diff --git a/plugins/scaffolder-backend/src/service/router.ts b/plugins/scaffolder-backend/src/service/router.ts index a72a5c5d48..23af538df8 100644 --- a/plugins/scaffolder-backend/src/service/router.ts +++ b/plugins/scaffolder-backend/src/service/router.ts @@ -34,17 +34,17 @@ import { CatalogEntityClient } from '../lib/catalog'; import { validate, ValidatorResult } from 'jsonschema'; import parseGitUrl from 'git-url-parse'; import { - MemoryTaskBroker, - MemoryDatabase, + DatabaseTaskStore, + StorageTaskBroker, TaskWorker, } from '../scaffolder/tasks'; import { TemplateActionRegistry, templateEntityToSpec, } from '../scaffolder/tasks/TemplateConverter'; -import { LOCATION_ANNOTATION } from '@backstage/catalog-model'; import { registerLegacyActions } from '../scaffolder/stages/legacy'; import { getWorkingDirectory } from './helpers'; +import { PluginDatabaseManager } from '@backstage/backend-common'; export interface RouterOptions { preparers: PreparerBuilder; @@ -55,6 +55,7 @@ export interface RouterOptions { config: Config; dockerClient: Docker; entityClient: CatalogEntityClient; + database: PluginDatabaseManager; } export async function createRouter( @@ -71,12 +72,17 @@ export async function createRouter( config, dockerClient, entityClient, + database, } = options; const logger = parentLogger.child({ plugin: 'scaffolder' }); const workingDirectory = await getWorkingDirectory(config, logger); const jobProcessor = await JobProcessor.fromConfig({ config, logger }); - const taskBroker = new MemoryTaskBroker(new MemoryDatabase()); + + const databaseTaskStore = await DatabaseTaskStore.create( + await database.getClient(), + ); + const taskBroker = new StorageTaskBroker(databaseTaskStore); const actionRegistry = new TemplateActionRegistry(); const worker = new TaskWorker({ logger,