From 1c1ff59ebac9a8b150da6d47a1e07ac1a8f860d8 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Fri, 22 Jan 2021 16:58:41 +0100 Subject: [PATCH] Add new apis using task broker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Fredrik Adelöw Co-authored-by: blam Co-authored-by: Patrik Oldsberg --- .../scaffolder/tasks/MemoryDatabase.test.ts | 21 ++++ .../src/scaffolder/tasks/MemoryDatabase.ts | 50 +++++++- .../src/scaffolder/tasks/MemoryTaskBroker.ts | 54 ++++++++- .../src/scaffolder/tasks/TaskWorker.ts | 110 ++++++++++++++++++ .../src/scaffolder/tasks/index.ts | 19 +++ .../src/scaffolder/tasks/types.ts | 11 +- .../scaffolder-backend/src/service/router.ts | 73 ++++++++++++ 7 files changed, 325 insertions(+), 13 deletions(-) create mode 100644 plugins/scaffolder-backend/src/scaffolder/tasks/MemoryDatabase.test.ts create mode 100644 plugins/scaffolder-backend/src/scaffolder/tasks/TaskWorker.ts create mode 100644 plugins/scaffolder-backend/src/scaffolder/tasks/index.ts diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/MemoryDatabase.test.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/MemoryDatabase.test.ts new file mode 100644 index 0000000000..c5522cb38a --- /dev/null +++ b/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); + }); +}); diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/MemoryDatabase.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/MemoryDatabase.ts index 65e64bf2a8..69fb6a1567 100644 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/MemoryDatabase.ts +++ b/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; createTask(task: TaskSpec): Promise; claimTask(): Promise; - heartBeat(runId: string): Promise; + heartbeat(runId: string): Promise; setStatus(taskId: string, status: Status): Promise; } +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(); + private readonly events = new Array(); - async heartBeat(runId: string): Promise { + 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 { 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(), }); } diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/MemoryTaskBroker.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/MemoryTaskBroker.ts index a5d90898f5..2491c9a091 100644 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/MemoryTaskBroker.ts +++ b/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 { - 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 { - 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; } diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/TaskWorker.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/TaskWorker.ts new file mode 100644 index 0000000000..dc15b3db65 --- /dev/null +++ b/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'); + } + } +} diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/index.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/index.ts new file mode 100644 index 0000000000..1cc9842fed --- /dev/null +++ b/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'; diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/types.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/types.ts index e8cea06005..12cb9fee1f 100644 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/types.ts +++ b/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 = { diff --git a/plugins/scaffolder-backend/src/service/router.ts b/plugins/scaffolder-backend/src/service/router.ts index 9098e5d05e..706027f62e 100644 --- a/plugins/scaffolder-backend/src/service/router.ts +++ b/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 = {