Merge pull request #4360 from backstage/mob/stateless-scaffolder

Introduce stateless v2 scaffolder API

.changeset/afraid-teachers-cross.md

@@ -0,0 +1,29 @@
+---
+'@backstage/create-app': patch
+---
+
+Pass on plugin database management instance that is now required by the scaffolder plugin.
+
+To apply this change to an existing application, add the following to `src/plugins/scaffolder.ts`:
+
+```diff
+export default async function createPlugin({
+  logger,
+  config,
++  database,
+}: PluginEnvironment) {
+
+// ...omitted...
+
+  return await createRouter({
+    preparers,
+    templaters,
+    publishers,
+    logger,
+    config,
+    dockerClient,
+    entityClient,
++    database,
+  });
+}
+```

.changeset/large-terms-invite.md

@@ -0,0 +1,7 @@
+---
+'@backstage/plugin-scaffolder-backend': minor
+---
+
+Introduced `v2` Scaffolder REST API, which uses an implementation that is database backed, making the scaffolder instances stateless. The `createRouter` function now requires a `PluginDatabaseManager` instance to be passed in, commonly available as `database` in the plugin environment in the backend.
+
+This API should be considered unstable until used by the scaffolder frontend.

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

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

plugins/scaffolder-backend/migrations/20210120143715_init.js

@@ -0,0 +1,85 @@
+/*
+ * 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.
+ */
+
+// @ts-check
+
+/**
+ * @param {import('knex')} knex
+ */
+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('spec')
+      .notNullable()
+      .comment('A JSON encoded task specification');
+    table
+      .text('status')
+      .notNullable()
+      .comment('The current status of the task');
+    table
+      .dateTime('created_at')
+      .defaultTo(knex.fn.now())
+      .notNullable()
+      .comment('The timestamp when this task was created');
+    table
+      .dateTime('last_heartbeat_at')
+      .nullable()
+      .comment('The last timestamp when a heartbeat was received');
+  });
+
+  await knex.schema.createTable('task_events', table => {
+    table.comment('The event stream a given task');
+    table
+      .bigIncrements('id')
+      .primary()
+      .notNullable()
+      .comment('The ID of the event');
+    table
+      .uuid('task_id')
+      .references('id')
+      .inTable('tasks')
+      .notNullable()
+      .onDelete('CASCADE')
+      .comment('The task that generated the event');
+    table
+      .text('body')
+      .notNullable()
+      .comment('The JSON encoded body of the event');
+    table.text('event_type').notNullable().comment('The type of event');
+    table
+      .timestamp('created_at')
+      .defaultTo(knex.fn.now())
+      .notNullable()
+      .comment('The timestamp when this event was generated');
+
+    table.index(['task_id'], 'task_events_task_id_idx');
+  });
+};
+
+/**
+ * @param {import('knex')} knex
+ */
+exports.down = async function down(knex) {
+  if (knex.client.config.client !== 'sqlite3') {
+    await knex.schema.alterTable('task_events', table => {
+      table.dropIndex([], 'ctask_events_task_id_idx');
+    });
+  }
+  await knex.schema.dropTable('task_events');
+  await knex.schema.dropTable('tasks');
+};

plugins/scaffolder-backend/package.json

@@ -53,6 +53,7 @@
     "helmet": "^4.0.0",
     "isomorphic-git": "^1.8.0",
     "jsonschema": "^1.2.6",
+    "knex": "^0.21.6",
     "morgan": "^1.10.0",
     "uuid": "^8.2.0",
     "winston": "^3.2.1",
@@ -71,6 +72,7 @@
   },
   "files": [
     "dist",
+    "migrations",
     "config.d.ts"
   ],
   "configSchema": "config.d.ts"

plugins/scaffolder-backend/src/scaffolder/stages/legacy.ts

@@ -0,0 +1,110 @@
+/*
+ * 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 { TemplateActionRegistry } from '../tasks/TemplateConverter';
+import { FilePreparer, PreparerBuilder } from './prepare';
+import Docker from 'dockerode';
+import { TemplaterBuilder, TemplaterValues } from './templater';
+import { PublisherBuilder } from './publish';
+
+type Options = {
+  dockerClient: Docker;
+  preparers: PreparerBuilder;
+  templaters: TemplaterBuilder;
+  publishers: PublisherBuilder;
+};
+
+export function registerLegacyActions(
+  registry: TemplateActionRegistry,
+  options: Options,
+) {
+  const { dockerClient, preparers, templaters, publishers } = options;
+
+  registry.register({
+    id: 'legacy:prepare',
+    async handler(ctx) {
+      const { protocol, url } = ctx.parameters;
+      const preparer =
+        protocol === 'file' ? new FilePreparer() : preparers.get(url as string);
+
+      ctx.logger.info('Prepare the skeleton');
+      await preparer.prepare({
+        url: url as string,
+        logger: ctx.logger,
+        workspacePath: ctx.workspacePath,
+      });
+    },
+  });
+
+  registry.register({
+    id: 'legacy:template',
+    async handler(ctx) {
+      const { logger } = ctx;
+
+      const templater = templaters.get(ctx.parameters.templater as string);
+
+      logger.info('Run the templater');
+      await templater.run({
+        workspacePath: ctx.workspacePath,
+        dockerClient,
+        logStream: ctx.logStream,
+        values: ctx.parameters.values as TemplaterValues,
+      });
+    },
+  });
+
+  registry.register({
+    id: 'legacy:publish',
+    async handler(ctx) {
+      const { values } = ctx.parameters;
+      if (
+        typeof values !== 'object' ||
+        values === null ||
+        Array.isArray(values)
+      ) {
+        throw new Error(
+          `Invalid values passed to publish, got ${typeof values}`,
+        );
+      }
+      const storePath = values.storePath as unknown;
+      if (typeof storePath !== 'string') {
+        throw new Error(
+          `Invalid store path passed to publish, got ${typeof storePath}`,
+        );
+      }
+      const owner = values.owner as unknown;
+      if (typeof owner !== 'string') {
+        throw new Error(`Invalid owner passed to publish, got ${typeof owner}`);
+      }
+
+      const publisher = publishers.get(storePath);
+      ctx.logger.info('Will now store the template');
+      const { remoteUrl, catalogInfoUrl } = await publisher.publish({
+        values: {
+          ...values,
+          owner,
+          storePath,
+        },
+        workspacePath: ctx.workspacePath,
+        logger: ctx.logger,
+      });
+      ctx.output('remoteUrl', remoteUrl);
+      if (catalogInfoUrl) {
+        ctx.output('catalogInfoUrl', catalogInfoUrl);
+      }
+    },
+  });
+}

plugins/scaffolder-backend/src/scaffolder/tasks/DatabaseTaskStore.ts

@@ -0,0 +1,272 @@
+/*
+ * 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 from 'knex';
+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;
+  created_at: string;
+};
+
+export type RawDbTaskEventRow = {
+  id: number;
+  task_id: string;
+  body: string;
+  event_type: TaskEventType;
+  created_at: string;
+};
+
+export class DatabaseTaskStore implements TaskStore {
+  static async create(knex: Knex): Promise<DatabaseTaskStore> {
+    await knex.migrate.latest({
+      directory: migrationsDir,
+    });
+    return new DatabaseTaskStore(knex);
+  }
+
+  constructor(private readonly db: Knex) {}
+
+  async get(taskId: string): Promise<DbTaskRow> {
+    const [result] = await this.db<RawDbTaskRow>('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,
+        lastHeartbeatAt: result.last_heartbeat_at,
+        createdAt: result.created_at,
+      };
+    } 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<RawDbTaskRow>('tasks').insert({
+      id: taskId,
+      spec: JSON.stringify(spec),
+      status: 'open',
+    });
+    return { taskId };
+  }
+
+  async claimTask(): Promise<DbTaskRow | undefined> {
+    return this.db.transaction(async tx => {
+      const [task] = await tx<RawDbTaskRow>('tasks')
+        .where({
+          status: 'open',
+        })
+        .limit(1)
+        .select();
+
+      if (!task) {
+        return undefined;
+      }
+
+      const updateCount = await tx<RawDbTaskRow>('tasks')
+        .where({ id: task.id, status: 'open' })
+        .update({
+          status: 'processing',
+          last_heartbeat_at: this.db.fn.now(),
+        });
+
+      if (updateCount < 1) {
+        return undefined;
+      }
+
+      try {
+        const spec = JSON.parse(task.spec);
+        return {
+          id: task.id,
+          spec,
+          status: 'processing',
+          lastHeartbeatAt: task.last_heartbeat_at,
+          createdAt: task.created_at,
+        };
+      } catch (error) {
+        throw new Error(`Failed to parse spec of task '${task.id}', ${error}`);
+      }
+    });
+  }
+
+  async heartbeatTask(taskId: string): Promise<void> {
+    const updateCount = await this.db<RawDbTaskRow>('tasks')
+      .where({ id: taskId, status: 'processing' })
+      .update({
+        last_heartbeat_at: this.db.fn.now(),
+      });
+    if (updateCount === 0) {
+      throw new ConflictError(`No running task with taskId ${taskId} found`);
+    }
+  }
+
+  async listStaleTasks({
+    timeoutS,
+  }: {
+    timeoutS: number;
+  }): Promise<{
+    tasks: { taskId: string }[];
+  }> {
+    const rawRows = await this.db<RawDbTaskRow>('tasks')
+      .where('status', 'processing')
+      .andWhere(
+        'last_heartbeat_at',
+        '<=',
+        this.db.client.config.client === 'sqlite3'
+          ? this.db.raw(`datetime('now', ?)`, [`-${timeoutS} seconds`])
+          : this.db.raw(`dateadd('second', ?, ?)`, [
+              `-${timeoutS}`,
+              this.db.fn.now(),
+            ]),
+      );
+    const tasks = rawRows.map(row => ({
+      taskId: row.id,
+    }));
+    return { tasks };
+  }
+
+  async completeTask({
+    taskId,
+    status,
+    eventBody,
+  }: {
+    taskId: string;
+    status: Status;
+    eventBody: JsonObject;
+  }): Promise<void> {
+    let oldStatus: string;
+    if (status === 'failed' || status === 'completed') {
+      oldStatus = 'processing';
+    } else {
+      throw new Error(
+        `Invalid status update of run '${taskId}' to status '${status}'`,
+      );
+    }
+    await this.db.transaction(async tx => {
+      const [task] = await tx<RawDbTaskRow>('tasks')
+        .where({
+          id: taskId,
+        })
+        .limit(1)
+        .select();
+
+      if (!task) {
+        throw new Error(`No task with taskId ${taskId} found`);
+      }
+      if (task.status !== oldStatus) {
+        throw new ConflictError(
+          `Refusing to update status of run '${taskId}' to status '${status}' ` +
+            `as it is currently '${task.status}', expected '${oldStatus}'`,
+        );
+      }
+      const updateCount = await tx<RawDbTaskRow>('tasks')
+        .where({
+          id: taskId,
+          status: oldStatus,
+        })
+        .update({
+          status,
+        });
+      if (updateCount !== 1) {
+        throw new ConflictError(
+          `Failed to update status to '${status}' for taskId ${taskId}`,
+        );
+      }
+
+      await tx<RawDbTaskEventRow>('task_events').insert({
+        task_id: taskId,
+        event_type: 'completion',
+        body: JSON.stringify(eventBody),
+      });
+    });
+  }
+
+  async emitLogEvent({ taskId, body }: TaskStoreEmitOptions): Promise<void> {
+    const serliazedBody = JSON.stringify(body);
+    await this.db<RawDbTaskEventRow>('task_events').insert({
+      task_id: taskId,
+      event_type: 'log',
+      body: serliazedBody,
+    });
+  }
+
+  async listEvents({
+    taskId,
+    after,
+  }: TaskStoreGetEventsOptions): Promise<{ events: DbTaskEventRow[] }> {
+    const rawEvents = await this.db<RawDbTaskEventRow>('task_events')
+      .where({
+        task_id: taskId,
+      })
+      .andWhere(builder => {
+        if (typeof after === 'number') {
+          builder.where('id', '>', after).orWhere('event_type', 'completion');
+        }
+      })
+      .orderBy('id')
+      .select();
+
+    const events = rawEvents.map(event => {
+      try {
+        const body = JSON.parse(event.body) as JsonObject;
+        return {
+          id: event.id,
+          taskId,
+          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 };
+  }
+}

plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.test.ts

@@ -0,0 +1,183 @@
+/*
+ * 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 {
+  getVoidLogger,
+  SingleConnectionDatabaseManager,
+} from '@backstage/backend-common';
+import { ConfigReader } from '@backstage/config';
+import { DatabaseTaskStore } from './DatabaseTaskStore';
+import { StorageTaskBroker, TaskAgent } from './StorageTaskBroker';
+import { TaskSpec, DbTaskEventRow } from './types';
+
+async function createStore(): Promise<DatabaseTaskStore> {
+  const manager = SingleConnectionDatabaseManager.fromConfig(
+    new ConfigReader({
+      backend: {
+        database: {
+          client: 'sqlite3',
+          connection: ':memory:',
+        },
+      },
+    }),
+  ).forPlugin('scaffolder');
+  return await DatabaseTaskStore.create(await manager.getClient());
+}
+
+describe('StorageTaskBroker', () => {
+  let storage: DatabaseTaskStore;
+
+  beforeAll(async () => {
+    storage = await createStore();
+  });
+
+  const logger = getVoidLogger();
+  it('should claim a dispatched work item', async () => {
+    const broker = new StorageTaskBroker(storage, logger);
+    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, logger);
+    const promise = broker.claim();
+
+    await expect(Promise.race([promise, 'waiting'])).resolves.toBe('waiting');
+
+    await broker.dispatch({ steps: [] });
+    await expect(promise).resolves.toEqual(expect.any(TaskAgent));
+  });
+
+  it('should dispatch multiple items and claim them in order', async () => {
+    const broker = new StorageTaskBroker(storage, logger);
+    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();
+    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.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 broker = new StorageTaskBroker(storage, logger);
+    const dispatchResult = await broker.dispatch({ steps: [] });
+    const task = await broker.claim();
+    await task.complete('completed');
+    const taskRow = await storage.get(dispatchResult.taskId);
+    expect(taskRow.status).toBe('completed');
+  }, 10000);
+
+  it('should fail a task', async () => {
+    const broker = new StorageTaskBroker(storage, logger);
+    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, logger);
+    const broker2 = new StorageTaskBroker(storage, logger);
+
+    const { taskId } = await broker1.dispatch({ steps: [] });
+
+    const logPromise = new Promise<DbTaskEventRow[]>(resolve => {
+      const observedEvents = new Array<DbTaskEventRow>();
+
+      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, logger)).toEqual([
+      'log 1',
+      'log 2',
+      'log 3',
+      'Run completed with status: completed',
+    ]);
+
+    const afterLogs = await new Promise<string[]>(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, logger);
+    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.lastHeartbeatAt !== initialTask.lastHeartbeatAt) {
+        break;
+      }
+      await new Promise(resolve => setTimeout(resolve, 50));
+    }
+    await task.complete('completed');
+    expect.assertions(0);
+  });
+
+  it('should be update the status to failed if heartbeat fails', async () => {
+    const broker = new StorageTaskBroker(storage, logger);
+    const { taskId } = await broker.dispatch({ steps: [] });
+    const task = await broker.claim();
+
+    jest
+      .spyOn((task as any).storage, 'heartbeatTask')
+      .mockRejectedValue(new Error('nah m8'));
+
+    const intervalId = setInterval(() => {
+      broker.vacuumTasks({ timeoutS: 2 }).catch(fail);
+    }, 500);
+
+    for (;;) {
+      const maybeTask = await storage.get(taskId);
+      if (maybeTask.status === 'failed') {
+        break;
+      }
+      await new Promise(resolve => setTimeout(resolve, 50));
+    }
+
+    clearInterval(intervalId);
+
+    expect(task.done).toBe(true);
+  });
+});

plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.ts

@@ -0,0 +1,205 @@
+/*
+ * 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 { Logger } from 'winston';
+import {
+  CompletedTaskState,
+  Task,
+  TaskSpec,
+  TaskStore,
+  TaskBroker,
+  DispatchResult,
+  DbTaskEventRow,
+} from './types';
+
+export class TaskAgent implements Task {
+  private isDone = false;
+
+  private heartbeatTimeoutId?: ReturnType<typeof setInterval>;
+
+  static create(state: TaskState, storage: TaskStore, logger: Logger) {
+    const agent = new TaskAgent(state, storage, logger);
+    agent.startTimeout();
+    return agent;
+  }
+
+  // Runs heartbeat internally
+  private constructor(
+    private readonly state: TaskState,
+    private readonly storage: TaskStore,
+    private readonly logger: Logger,
+  ) {}
+
+  get spec() {
+    return this.state.spec;
+  }
+
+  async getWorkspaceName() {
+    return this.state.taskId;
+  }
+
+  get done() {
+    return this.isDone;
+  }
+
+  async emitLog(message: string): Promise<void> {
+    await this.storage.emitLogEvent({
+      taskId: this.state.taskId,
+      body: { message },
+    });
+  }
+
+  async complete(result: CompletedTaskState): Promise<void> {
+    await this.storage.completeTask({
+      taskId: this.state.taskId,
+      status: result === 'failed' ? 'failed' : 'completed',
+      eventBody: { message: `Run completed with status: ${result}` },
+    });
+    this.isDone = true;
+    if (this.heartbeatTimeoutId) {
+      clearTimeout(this.heartbeatTimeoutId);
+    }
+  }
+
+  private startTimeout() {
+    this.heartbeatTimeoutId = setTimeout(async () => {
+      try {
+        await this.storage.heartbeatTask(this.state.taskId);
+        this.startTimeout();
+      } catch (error) {
+        this.isDone = true;
+
+        this.logger.error(
+          `Heartbeat for task ${this.state.taskId} failed`,
+          error,
+        );
+      }
+    }, 1000);
+  }
+}
+
+interface TaskState {
+  spec: TaskSpec;
+  taskId: string;
+}
+
+function defer() {
+  let resolve = () => {};
+  const promise = new Promise<void>(_resolve => {
+    resolve = _resolve;
+  });
+  return { promise, resolve };
+}
+
+export class StorageTaskBroker implements TaskBroker {
+  constructor(
+    private readonly storage: TaskStore,
+    private readonly logger: Logger,
+  ) {}
+  private deferredDispatch = defer();
+
+  async claim(): Promise<Task> {
+    for (;;) {
+      const pendingTask = await this.storage.claimTask();
+      if (pendingTask) {
+        return TaskAgent.create(
+          {
+            taskId: pendingTask.id,
+            spec: pendingTask.spec,
+          },
+          this.storage,
+          this.logger,
+        );
+      }
+
+      await this.waitForDispatch();
+    }
+  }
+
+  async dispatch(spec: TaskSpec): Promise<DispatchResult> {
+    const taskRow = await this.storage.createTask(spec);
+    this.signalDispatch();
+    return {
+      taskId: taskRow.taskId,
+    };
+  }
+
+  observe(
+    options: {
+      taskId: string;
+      after: number | undefined;
+    },
+    callback: (
+      error: Error | undefined,
+      result: { events: DbTaskEventRow[] },
+    ) => void,
+  ): () => void {
+    const { taskId } = options;
+
+    let cancelled = false;
+    const unsubscribe = () => {
+      cancelled = true;
+    };
+
+    (async () => {
+      let after = options.after;
+      while (!cancelled) {
+        const result = await this.storage.listEvents({ taskId, after: after });
+        const { events } = result;
+        if (events.length) {
+          after = events[events.length - 1].id;
+          try {
+            callback(undefined, result);
+          } catch (error) {
+            callback(error, { events: [] });
+          }
+        }
+
+        await new Promise(resolve => setTimeout(resolve, 1000));
+      }
+    })();
+
+    return unsubscribe;
+  }
+
+  async vacuumTasks(timeoutS: { timeoutS: number }): Promise<void> {
+    const { tasks } = await this.storage.listStaleTasks(timeoutS);
+    await Promise.all(
+      tasks.map(async task => {
+        try {
+          await this.storage.completeTask({
+            taskId: task.taskId,
+            status: 'failed',
+            eventBody: {
+              message:
+                'The task was cancelled because the task worker lost connection to the task broker',
+            },
+          });
+        } catch (error) {
+          this.logger.warn(`Failed to cancel task '${task.taskId}', ${error}`);
+        }
+      }),
+    );
+  }
+
+  private waitForDispatch() {
+    return this.deferredDispatch.promise;
+  }
+
+  private signalDispatch() {
+    this.deferredDispatch.resolve();
+    this.deferredDispatch = defer();
+  }
+}

plugins/scaffolder-backend/src/scaffolder/tasks/TaskWorker.ts

@@ -0,0 +1,110 @@
+/*
+ * 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 { PassThrough } from 'stream';
+import { Logger } from 'winston';
+import * as winston from 'winston';
+import { JsonValue } from '@backstage/config';
+import { TaskBroker, Task } from './types';
+import fs from 'fs-extra';
+import path from 'path';
+import { TemplateActionRegistry } from './TemplateConverter';
+
+type Options = {
+  logger: Logger;
+  taskBroker: TaskBroker;
+  workingDirectory: string;
+  actionRegistry: TemplateActionRegistry;
+};
+
+export class TaskWorker {
+  constructor(private readonly options: Options) {}
+
+  start() {
+    (async () => {
+      for (;;) {
+        const task = await this.options.taskBroker.claim();
+        await this.runOneTask(task);
+      }
+    })();
+  }
+
+  async runOneTask(task: Task) {
+    try {
+      const { actionRegistry, logger } = this.options;
+
+      const workspacePath = path.join(
+        this.options.workingDirectory,
+        await task.getWorkspaceName(),
+      );
+      await fs.ensureDir(workspacePath);
+
+      const taskLogger = winston.createLogger({
+        level: process.env.LOG_LEVEL || 'info',
+        format: winston.format.combine(
+          winston.format.colorize(),
+          winston.format.timestamp(),
+          winston.format.simple(),
+        ),
+        defaultMeta: {},
+      });
+
+      const stream = new PassThrough();
+      stream.on('data', data => {
+        const message = data.toString().trim();
+        if (message?.length > 1) task.emitLog(message);
+      });
+
+      taskLogger.add(new winston.transports.Stream({ stream }));
+
+      // Give us some time to curl observe
+      task.emitLog('Task claimed, waiting ...');
+      await new Promise(resolve => setTimeout(resolve, 5000));
+      task.emitLog(`Starting up work with ${task.spec.steps.length} steps`);
+
+      const outputs: { [name: string]: JsonValue } = {};
+
+      for (const step of task.spec.steps) {
+        task.emitLog(`Beginning step ${step.name}`);
+
+        const action = actionRegistry.get(step.action);
+        if (!action) {
+          throw new Error(`Action '${step.action}' does not exist`);
+        }
+
+        // TODO: substitute any placeholders with output from previous steps
+        const parameters = step.parameters!;
+
+        await action.handler({
+          logger,
+          logStream: stream,
+          parameters,
+          workspacePath,
+          output(name: string, value: JsonValue) {
+            outputs[name] = value;
+          },
+        });
+
+        task.emitLog(`Finished step ${step.name}`);
+      }
+
+      await task.complete('completed');
+    } catch (error) {
+      task.emitLog(String(error.stack));
+      await task.complete('failed');
+    }
+  }
+}

plugins/scaffolder-backend/src/scaffolder/tasks/TemplateConverter.ts

@@ -0,0 +1,117 @@
+/*
+ * 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 { resolve as resolvePath } from 'path';
+import { JsonValue } from '@backstage/config';
+import { TemplateEntityV1alpha1 } from '@backstage/catalog-model';
+import { Logger } from 'winston';
+import type { Writable } from 'stream';
+
+import { TaskSpec } from './types';
+import { ConflictError, NotFoundError } from '@backstage/backend-common';
+import {
+  getTemplaterKey,
+  joinGitUrlPath,
+  parseLocationAnnotation,
+  TemplaterValues,
+} from '../stages';
+
+export function templateEntityToSpec(
+  template: TemplateEntityV1alpha1,
+  values: TemplaterValues,
+): TaskSpec {
+  const steps: TaskSpec['steps'] = [];
+
+  const { protocol, location } = parseLocationAnnotation(template);
+
+  let url: string;
+  if (protocol === 'file') {
+    const path = resolvePath(location, template.spec.path || '.');
+
+    url = `file://${path}`;
+  } else {
+    url = joinGitUrlPath(location, template.spec.path);
+  }
+  const templater = getTemplaterKey(template);
+
+  steps.push({
+    id: 'prepare',
+    name: 'Prepare',
+    action: 'legacy:prepare',
+    parameters: {
+      protocol,
+      url,
+    },
+  });
+
+  steps.push({
+    id: 'template',
+    name: 'Template',
+    action: 'legacy:template',
+    parameters: {
+      templater,
+      values,
+    },
+  });
+
+  steps.push({
+    id: 'publish',
+    name: 'Publishing',
+    action: 'legacy:publish',
+    parameters: {
+      values,
+    },
+  });
+
+  return { steps };
+}
+
+type ActionContext = {
+  logger: Logger;
+  logStream: Writable;
+
+  workspacePath: string;
+  parameters: { [name: string]: JsonValue };
+  output(name: string, value: JsonValue): void;
+};
+
+type TemplateAction = {
+  id: string;
+  handler: (ctx: ActionContext) => Promise<void>;
+};
+
+export class TemplateActionRegistry {
+  private readonly actions = new Map<string, TemplateAction>();
+
+  register(action: TemplateAction) {
+    if (this.actions.has(action.id)) {
+      throw new ConflictError(
+        `Template action with ID '${action.id}' has already been registered`,
+      );
+    }
+    this.actions.set(action.id, action);
+  }
+
+  get(actionId: string): TemplateAction {
+    const action = this.actions.get(actionId);
+    if (!action) {
+      throw new NotFoundError(
+        `Template action with ID '${actionId}' is not registered.`,
+      );
+    }
+    return action;
+  }
+}

plugins/scaffolder-backend/src/scaffolder/tasks/index.ts

@@ -0,0 +1,19 @@
+/*
+ * 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.
+ */
+
+export { DatabaseTaskStore } from './DatabaseTaskStore';
+export { StorageTaskBroker } from './StorageTaskBroker';
+export { TaskWorker } from './TaskWorker';

plugins/scaffolder-backend/src/scaffolder/tasks/types.ts

@@ -0,0 +1,111 @@
+/*
+ * 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 { JsonValue, JsonObject } from '@backstage/config';
+
+export type Status =
+  | 'open'
+  | 'processing'
+  | 'failed'
+  | 'cancelled'
+  | 'completed';
+
+export type CompletedTaskState = 'failed' | 'completed';
+
+export type DbTaskRow = {
+  id: string;
+  spec: TaskSpec;
+  status: Status;
+  createdAt: string;
+  lastHeartbeatAt?: string;
+};
+
+export type TaskEventType = 'completion' | 'log';
+export type DbTaskEventRow = {
+  id: number;
+  taskId: string;
+  body: JsonObject;
+  type: TaskEventType;
+  createdAt: string;
+};
+
+export type TaskSpec = {
+  steps: Array<{
+    id: string;
+    name: string;
+    action: string;
+    parameters?: { [name: string]: JsonValue };
+  }>;
+};
+
+export type DispatchResult = {
+  taskId: string;
+};
+
+export interface Task {
+  spec: TaskSpec;
+  done: boolean;
+  emitLog(message: string): Promise<void>;
+  complete(result: CompletedTaskState): Promise<void>;
+  getWorkspaceName(): Promise<string>;
+}
+
+export interface TaskBroker {
+  claim(): Promise<Task>;
+  dispatch(spec: TaskSpec): Promise<DispatchResult>;
+  vacuumTasks(timeoutS: { timeoutS: number }): Promise<void>;
+  observe(
+    options: {
+      taskId: string;
+      after: number | undefined;
+    },
+    callback: (
+      error: Error | undefined,
+      result: { events: DbTaskEventRow[] },
+    ) => void,
+  ): () => void;
+}
+
+export type TaskStoreEmitOptions = {
+  taskId: string;
+  body: JsonObject;
+};
+
+export type TaskStoreGetEventsOptions = {
+  taskId: string;
+  after?: number | undefined;
+};
+export interface TaskStore {
+  createTask(task: TaskSpec): Promise<{ taskId: string }>;
+  claimTask(): Promise<DbTaskRow | undefined>;
+  completeTask(options: {
+    taskId: string;
+    status: Status;
+    eventBody: JsonObject;
+  }): Promise<void>;
+  heartbeatTask(taskId: string): Promise<void>;
+  listStaleTasks(options: {
+    timeoutS: number;
+  }): Promise<{
+    tasks: { taskId: string }[];
+  }>;
+
+  emitLogEvent({ taskId, body }: TaskStoreEmitOptions): Promise<void>;
+  listEvents({
+    taskId,
+    after,
+  }: TaskStoreGetEventsOptions): Promise<{ events: DbTaskEventRow[] }>;
+}

plugins/scaffolder-backend/src/service/helpers.ts

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2020 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 os from 'os';
+import fs from 'fs-extra';
+import { Logger } from 'winston';
+import { Config } from '@backstage/config';
+
+export async function getWorkingDirectory(
+  config: Config,
+  logger: Logger,
+): Promise<string> {
+  if (!config.has('backend.workingDirectory')) {
+    return os.tmpdir();
+  }
+
+  const workingDirectory = config.getString('backend.workingDirectory');
+  try {
+    // Check if working directory exists and is writable
+    await fs.access(workingDirectory, fs.constants.F_OK | fs.constants.W_OK);
+    logger.info(`using working directory: ${workingDirectory}`);
+  } catch (err) {
+    logger.error(
+      `working directory ${workingDirectory} ${
+        err.code === 'ENOENT' ? 'does not exist' : 'is not writable'
+      }`,
+    );
+    throw err;
+  }
+  return workingDirectory;
+}

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

plugins/scaffolder-backend/src/service/router.ts

@@ -33,6 +33,18 @@ import {
 import { CatalogEntityClient } from '../lib/catalog';
 import { validate, ValidatorResult } from 'jsonschema';
 import parseGitUrl from 'git-url-parse';
+import {
+  DatabaseTaskStore,
+  StorageTaskBroker,
+  TaskWorker,
+} from '../scaffolder/tasks';
+import {
+  TemplateActionRegistry,
+  templateEntityToSpec,
+} from '../scaffolder/tasks/TemplateConverter';
+import { registerLegacyActions } from '../scaffolder/stages/legacy';
+import { getWorkingDirectory } from './helpers';
+import { PluginDatabaseManager } from '@backstage/backend-common';
 
 export interface RouterOptions {
   preparers: PreparerBuilder;
@@ -43,6 +55,7 @@ export interface RouterOptions {
   config: Config;
   dockerClient: Docker;
   entityClient: CatalogEntityClient;
+  database: PluginDatabaseManager;
 }
 
 export async function createRouter(
@@ -59,11 +72,34 @@ 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 databaseTaskStore = await DatabaseTaskStore.create(
+    await database.getClient(),
+  );
+  const taskBroker = new StorageTaskBroker(databaseTaskStore, logger);
+  const actionRegistry = new TemplateActionRegistry();
+  const worker = new TaskWorker({
+    logger,
+    taskBroker,
+    actionRegistry,
+    workingDirectory,
+  });
+
+  registerLegacyActions(actionRegistry, {
+    dockerClient,
+    preparers,
+    publishers,
+    templaters,
+  });
+
+  worker.start();
+
   router
     .get('/v1/job/:jobId', ({ params }, res) => {
       const job = jobProcessor.get(params.jobId);
@@ -184,6 +220,75 @@ export async function createRouter(
       res.status(201).json({ id: job.id });
     });
 
+  // NOTE: The v2 API is unstable
+  router
+    .post('/v2/tasks', async (req, res) => {
+      const templateName: string = req.body.templateName;
+      const values: TemplaterValues = {
+        ...req.body.values,
+        destination: {
+          git: parseGitUrl(req.body.values.storePath),
+        },
+      };
+      const template = await entityClient.findTemplate(templateName);
+
+      const validationResult: ValidatorResult = validate(
+        values,
+        template.spec.schema,
+      );
+
+      if (!validationResult.valid) {
+        res.status(400).json({ errors: validationResult.errors });
+        return;
+      }
+      const taskSpec = templateEntityToSpec(template, values);
+      const result = await taskBroker.dispatch(taskSpec);
+
+      res.status(201).json({ id: result.taskId });
+    })
+    .get('/v2/tasks/:taskId/eventstream', async (req, res) => {
+      const { taskId } = req.params;
+      const after = Number(req.query.after) || undefined;
+      logger.debug(`Event stream observing taskId '${taskId}' opened`);
+
+      // Mandatory headers and http status to keep connection open
+      res.writeHead(200, {
+        Connection: 'keep-alive',
+        'Cache-Control': 'no-cache',
+        'Content-Type': 'text/event-stream',
+      });
+
+      // After client opens connection send all events as string
+      const unsubscribe = taskBroker.observe(
+        { taskId, after },
+        (error, { events }) => {
+          if (error) {
+            logger.error(
+              `Received error from event stream when observing taskId '${taskId}', ${error}`,
+            );
+          }
+
+          for (const event of events) {
+            res.write(
+              `event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`,
+            );
+            if (event.type === 'completion') {
+              unsubscribe();
+              // Closing the event stream here would cause the frontend
+              // to automatically reconnect because it lost connection.
+            }
+          }
+          res.flush();
+        },
+      );
+      // When client closes connection we update the clients list
+      // avoiding the disconnected one
+      req.on('close', () => {
+        unsubscribe();
+        logger.debug(`Event stream observing taskId '${taskId}' closed`);
+      });
+    });
+
   const app = express();
   app.set('logger', logger);
   app.use('/', router);