Wire up database and rename database classes

Co-authored-by: Fredrik Adelöw <freben@gmail.com>
Co-authored-by: blam<ben@blam.sh>
Co-authored-by: Patrik Oldsberg <poldsberg@gmail.com>

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

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

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<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,
+        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<RawDbTaskRow>('tasks').insert({
+      id: taskId,
+      spec: JSON.stringify(spec),
+      status: 'open',
+      retry_count: 0,
+    });
+    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 runId = uuid();
+      const updateCount = await tx<RawDbTaskRow>('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<void> {
+    const updateCount = await this.db<RawDbTaskRow>('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<void> {
+    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<RawDbTaskRow>('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<RawDbTaskRow>('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<void> {
+    const serliazedBody = JSON.stringify(body);
+    await this.db<RawDbTaskEventRow>('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<RawDbTaskEventRow>('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 };
+  }
+}

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<DbTaskRow>;
-  createTask(task: TaskSpec): Promise<DbTaskRow>;
-  claimTask(): Promise<DbTaskRow | undefined>;
-  heartbeat(runId: string): Promise<void>;
-  setStatus(taskId: string, status: Status): Promise<void>;
-}
-
-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<string, DbTaskRow>();
   private readonly events = new Array<DbTaskEventRow>();
 
-  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<DbTaskRow> {
+  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<DbTaskRow> {
@@ -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 });
   }
 }

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,

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<typeof setInterval>;
 
-  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<void> {
     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<Task> {
@@ -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,

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

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

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<Task>;
   dispatch(spec: TaskSpec): Promise<DispatchResult>;
 }
+
+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<DbTaskRow>;
+  createTask(task: TaskSpec): Promise<{ taskId: string }>;
+  claimTask(): Promise<DbTaskRow | undefined>;
+  heartbeat(runId: string): Promise<void>;
+  setStatus(runId: string, status: Status): Promise<void>;
+  emit({ taskId, runId, body, type }: TaskStoreEmitOptions): Promise<void>;
+  getEvents({
+    taskId,
+    after,
+  }: TaskStoreGetEventsOptions): Promise<{ events: DbTaskEventRow[] }>;
+}

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,