Add new apis using task broker

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

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

@@ -0,0 +1,21 @@
+/*
+ * 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.
+ */
+
+describe('MemoryDatabase', () => {
+  it('should be tested', async () => {
+    expect(1).toBe(2);
+  });
+});

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

@@ -13,22 +13,62 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-import { DbTaskRow, Status, TaskSpec } from './types';
+
+import { DbTaskRow, DbTaskEventRow, Status, TaskSpec } from './types';
 import { v4 as uuid } from 'uuid';
-import { DateTime } from 'luxon';
 
 export interface Database {
   get(taskId: string): Promise<DbTaskRow>;
   createTask(task: TaskSpec): Promise<DbTaskRow>;
   claimTask(): Promise<DbTaskRow | undefined>;
-  heartBeat(runId: string): Promise<void>;
+  heartbeat(runId: string): Promise<void>;
   setStatus(taskId: string, status: Status): Promise<void>;
 }
 
+type EmitOptions = {
+  taskId: string;
+  runId: string;
+  event: string;
+};
+
+type ReadOptions = {
+  taskId: string;
+  after?: number | undefined;
+};
+
 export class MemoryDatabase implements Database {
   private readonly store = new Map<string, DbTaskRow>();
+  private readonly events = new Array<DbTaskEventRow>();
 
-  async heartBeat(runId: string): Promise<void> {
+  async emit({ taskId, runId, event }: EmitOptions) {
+    this.events.push({
+      id: this.events.length,
+      taskId,
+      runId,
+      event,
+      createdAt: new Date().toISOString(),
+    });
+  }
+
+  async getEvents({
+    taskId,
+    after,
+  }: ReadOptions): Promise<{ events: DbTaskEventRow[] }> {
+    const events = this.events.filter(event => {
+      if (event.taskId !== taskId) {
+        return false;
+      }
+      if (after !== undefined) {
+        if (event.id <= after) {
+          return false;
+        }
+      }
+      return true;
+    });
+    return { events };
+  }
+
+  async heartbeat(runId: string): Promise<void> {
     let task: DbTaskRow | undefined;
 
     for (const t of this.store.values()) {
@@ -43,7 +83,7 @@ export class MemoryDatabase implements Database {
 
     this.store.set(task.taskId, {
       ...task,
-      lastHeartbeat: DateTime.local().toString(),
+      lastHeartbeat: new Date().toISOString(),
     });
   }
 

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

@@ -20,6 +20,7 @@ import {
   TaskSpec,
   TaskBroker,
   DispatchResult,
+  DbTaskEventRow,
 } from './types';
 import { MemoryDatabase } from './MemoryDatabase';
 
@@ -43,14 +44,22 @@ export class TaskAgent implements Task {
   }
 
   async emitLog(message: string): Promise<void> {
-    throw new Error('Method not implemented.');
+    await this.storage.emit({
+      taskId: this.state.taskId,
+      runId: this.state.runId,
+      event: message,
+    });
   }
 
   async complete(result: CompletedTaskState): Promise<void> {
-    this.storage.setStatus(
+    await this.storage.setStatus(
       this.state.taskId,
       result === 'FAILED' ? 'FAILED' : 'COMPLETED',
     );
+
+    if (this.heartbeartInterval) {
+      clearInterval(this.heartbeartInterval);
+    }
   }
 
   private start() {
@@ -58,7 +67,7 @@ export class TaskAgent implements Task {
       if (!this.state.runId) {
         throw new Error('no run id provided');
       }
-      this.storage.heartBeat(this.state.runId);
+      this.storage.heartbeat(this.state.runId);
     }, 1000);
   }
 }
@@ -66,7 +75,7 @@ export class TaskAgent implements Task {
 interface TaskState {
   spec: TaskSpec;
   taskId: string;
-  runId: string | undefined;
+  runId: string;
 }
 
 function defer() {
@@ -87,7 +96,7 @@ export class MemoryTaskBroker implements TaskBroker {
       if (pendingTask) {
         return TaskAgent.create(
           {
-            runId: pendingTask.runId,
+            runId: pendingTask.runId!,
             taskId: pendingTask.taskId,
             spec: pendingTask.spec,
           },
@@ -107,6 +116,41 @@ export class MemoryTaskBroker implements TaskBroker {
     };
   }
 
+  observe(
+    options: {
+      taskId: string;
+      after: number | undefined;
+    },
+    callback: (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.getEvents({ taskId, after: after });
+        const { events } = result;
+        if (events.length) {
+          after = events[events.length - 1].id;
+          try {
+            callback(result);
+          } catch (error) {
+            console.log('DEBUG: error =', error);
+          }
+        }
+
+        await new Promise(resolve => setTimeout(resolve, 1000));
+      }
+    })();
+
+    return unsubscribe;
+  }
+
   private waitForDispatch() {
     return this.deferredDispatch.promise;
   }

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 { TaskBroker, Task } from './types';
+import { Logger } from 'winston';
+import Docker from 'dockerode';
+import { CatalogEntityClient } from '../../lib/catalog';
+import {
+  FilePreparer,
+  parseLocationAnnotation,
+  PreparerBuilder,
+  TemplaterBuilder,
+  PublisherBuilder,
+} from '../stages';
+
+type Options = {
+  logger: Logger;
+  taskBroker: TaskBroker;
+  workingDirectory: string;
+  dockerClient: Docker;
+  entityClient: CatalogEntityClient;
+  preparers: PreparerBuilder;
+  templaters: TemplaterBuilder;
+  publishers: PublisherBuilder;
+};
+
+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) {
+    const {
+      dockerClient,
+      preparers,
+      templaters,
+      publishers,
+      workingDirectory,
+      logger,
+    } = this.options;
+
+    try {
+      task.emitLog('Task claimed, waiting ...');
+      // Give us some time to curl observe
+      await new Promise(resolve => setTimeout(resolve, 5000));
+
+      const { values, template } = task.spec;
+      task.emitLog('Prepare the skeleton');
+      const { protocol, location: pullPath } = parseLocationAnnotation(
+        task.spec.template,
+      );
+
+      const preparer =
+        protocol === 'file' ? new FilePreparer() : preparers.get(pullPath);
+      const templater = templaters.get(template);
+      const publisher = publishers.get(values.storePath);
+
+      const skeletonDir = await preparer.prepare(task.spec.template, {
+        logger,
+        workingDirectory: workingDirectory,
+      });
+
+      task.emitLog('Run the templater');
+      const { resultDir } = await templater.run({
+        directory: skeletonDir,
+        dockerClient,
+        logStream: process.stdout, // yay
+        values: values,
+      });
+
+      task.emitLog('Publish template');
+      logger.info('Will now store the template');
+
+      logger.info('Totally storing the template now');
+      await new Promise(resolve => setTimeout(resolve, 5000));
+      // const result = await publisher.publish({
+      //   values: values,
+      //   directory: resultDir,
+      //   logger,
+      // });
+      // task.emitLog(`Result: ${JSON.stringify(result)}`);
+
+      task.emitLog(`Completely done now!`);
+
+      await task.complete('COMPLETED');
+    } catch (error) {
+      await task.complete('FAILED');
+    }
+  }
+}

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 { MemoryDatabase } from './MemoryDatabase';
+export { MemoryTaskBroker } from './MemoryTaskBroker';
+export { TaskWorker } from './TaskWorker';

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

@@ -14,6 +14,9 @@
  * limitations under the License.
  */
 
+import { TemplateEntityV1alpha1 } from '@backstage/catalog-model';
+import { TemplaterValues } from '..';
+
 export type Status =
   | 'OPEN'
   | 'PROCESSING'
@@ -35,13 +38,15 @@ export type DbTaskRow = {
 
 export type DbTaskEventRow = {
   id: number;
-  runId?: number;
-  stageName: string;
+  runId: string;
+  taskId: string;
+  event: string;
   createdAt: string;
 };
 
 export type TaskSpec = {
-  metadata: string;
+  template: TemplateEntityV1alpha1;
+  values: TemplaterValues;
 };
 
 export type DispatchResult = {

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

@@ -33,6 +33,11 @@ import {
 import { CatalogEntityClient } from '../lib/catalog';
 import { validate, ValidatorResult } from 'jsonschema';
 import parseGitUrl from 'git-url-parse';
+import {
+  MemoryTaskBroker,
+  MemoryDatabase,
+  TaskWorker,
+} from '../scaffolder/tasks';
 
 export interface RouterOptions {
   preparers: PreparerBuilder;
@@ -63,6 +68,18 @@ export async function createRouter(
 
   const logger = parentLogger.child({ plugin: 'scaffolder' });
   const jobProcessor = await JobProcessor.fromConfig({ config, logger });
+  const taskBroker = new MemoryTaskBroker(new MemoryDatabase());
+  const worker = new TaskWorker({
+    logger,
+    taskBroker,
+    workingDirectory: 'todo',
+    dockerClient,
+    entityClient,
+    preparers,
+    publishers,
+    templaters,
+  });
+  worker.start();
 
   router
     .get('/v1/job/:jobId', ({ params }, res) => {
@@ -88,6 +105,62 @@ export async function createRouter(
         error: job.error,
       });
     })
+    // curl -X POST -d '{"templateName":"springboot-template","values": {"storePath":"https://github.com/jhaals/foo", "component_id":"woop", "description": "apa", "owner": "me" }}' -H 'Content-Type: application/json' localhost:7000/api/scaffolder/v2/tasks
+    .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 result = await taskBroker.dispatch({
+        template,
+        values,
+      });
+
+      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.info('event stream 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 nests as string
+      const unsubscribe = taskBroker.observe(
+        { taskId, after },
+        ({ events }) => {
+          for (const event of events) {
+            res.write(`event:${JSON.stringify(event)}\n\n`);
+          }
+        },
+      );
+      // When client closes connection we update the clients list
+      // avoiding the disconnected one
+      req.on('close', () => {
+        unsubscribe();
+        logger.info('event stream closed');
+      });
+    })
+
     .post('/v1/jobs', async (req, res) => {
       const templateName: string = req.body.templateName;
       const values: TemplaterValues = {