From 9ef2e68b3399f3f331f4a66decd0e726480f0820 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Mon, 8 Feb 2021 14:08:13 +0100 Subject: [PATCH] Add log metadata and getTask method --- .../src/scaffolder/tasks/DatabaseTaskStore.ts | 2 +- .../tasks/StorageTaskBroker.test.ts | 10 +-- .../src/scaffolder/tasks/StorageTaskBroker.ts | 10 ++- .../src/scaffolder/tasks/TaskWorker.ts | 89 ++++++++++--------- .../src/scaffolder/tasks/types.ts | 3 +- 5 files changed, 61 insertions(+), 53 deletions(-) diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/DatabaseTaskStore.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/DatabaseTaskStore.ts index 5717f3c8c1..4190a5d58f 100644 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/DatabaseTaskStore.ts +++ b/plugins/scaffolder-backend/src/scaffolder/tasks/DatabaseTaskStore.ts @@ -64,7 +64,7 @@ export class DatabaseTaskStore implements TaskStore { constructor(private readonly db: Knex) {} - async get(taskId: string): Promise { + async getTask(taskId: string): Promise { const [result] = await this.db('tasks') .where({ id: taskId }) .select(); diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.test.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.test.ts index c45fa46335..ffb8de6e3d 100644 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.test.ts +++ b/plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.test.ts @@ -83,7 +83,7 @@ describe('StorageTaskBroker', () => { const dispatchResult = await broker.dispatch({ steps: [] }); const task = await broker.claim(); await task.complete('completed'); - const taskRow = await storage.get(dispatchResult.taskId); + const taskRow = await storage.getTask(dispatchResult.taskId); expect(taskRow.status).toBe('completed'); }, 10000); @@ -92,7 +92,7 @@ describe('StorageTaskBroker', () => { const dispatchResult = await broker.dispatch({ steps: [] }); const task = await broker.claim(); await task.complete('failed'); - const taskRow = await storage.get(dispatchResult.taskId); + const taskRow = await storage.getTask(dispatchResult.taskId); expect(taskRow.status).toBe('failed'); }); @@ -142,10 +142,10 @@ describe('StorageTaskBroker', () => { const { taskId } = await broker.dispatch({ steps: [] }); const task = await broker.claim(); - const initialTask = await storage.get(taskId); + const initialTask = await storage.getTask(taskId); for (;;) { - const maybeTask = await storage.get(taskId); + const maybeTask = await storage.getTask(taskId); if (maybeTask.lastHeartbeatAt !== initialTask.lastHeartbeatAt) { break; } @@ -169,7 +169,7 @@ describe('StorageTaskBroker', () => { }, 500); for (;;) { - const maybeTask = await storage.get(taskId); + const maybeTask = await storage.getTask(taskId); if (maybeTask.status === 'failed') { break; } diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.ts index 48c0d5e5ba..d3b2c098c7 100644 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.ts +++ b/plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.ts @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ +import { JsonObject } from '@backstage/config'; import { Logger } from 'winston'; import { CompletedTaskState, @@ -22,6 +23,7 @@ import { TaskBroker, DispatchResult, DbTaskEventRow, + DbTaskRow, } from './types'; export class TaskAgent implements Task { @@ -54,10 +56,10 @@ export class TaskAgent implements Task { return this.isDone; } - async emitLog(message: string): Promise { + async emitLog(message: string, metadata?: JsonObject): Promise { await this.storage.emitLogEvent({ taskId: this.state.taskId, - body: { message }, + body: { message, ...metadata }, }); } @@ -136,6 +138,10 @@ export class StorageTaskBroker implements TaskBroker { }; } + async get(taskId: string): Promise { + return this.storage.getTask(taskId); + } + observe( options: { taskId: string; diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/TaskWorker.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/TaskWorker.ts index c62495ca4c..d2f3b88e65 100644 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/TaskWorker.ts +++ b/plugins/scaffolder-backend/src/scaffolder/tasks/TaskWorker.ts @@ -51,59 +51,60 @@ export class TaskWorker { 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`); + await 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 metadata = { stepId: step.id }; + try { + 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 action = actionRegistry.get(step.action); - if (!action) { - throw new Error(`Action '${step.action}' does not exist`); + const stream = new PassThrough(); + stream.on('data', data => { + const message = data.toString().trim(); + if (message?.length > 1) task.emitLog(message, metadata); + }); + + taskLogger.add(new winston.transports.Stream({ stream })); + await task.emitLog(`Beginning step ${step.name}`, metadata); + + 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; + }, + }); + + await task.emitLog(`Finished step ${step.name}`, metadata); + } catch (error) { + await task.emitLog(String(error.stack), metadata); + throw error; } - - // 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'); } } diff --git a/plugins/scaffolder-backend/src/scaffolder/tasks/types.ts b/plugins/scaffolder-backend/src/scaffolder/tasks/types.ts index 0c2592109b..783528dbdf 100644 --- a/plugins/scaffolder-backend/src/scaffolder/tasks/types.ts +++ b/plugins/scaffolder-backend/src/scaffolder/tasks/types.ts @@ -58,7 +58,7 @@ export type DispatchResult = { export interface Task { spec: TaskSpec; done: boolean; - emitLog(message: string): Promise; + emitLog(message: string, metadata?: JsonValue): Promise; complete(result: CompletedTaskState): Promise; getWorkspaceName(): Promise; } @@ -90,6 +90,7 @@ export type TaskStoreGetEventsOptions = { }; export interface TaskStore { createTask(task: TaskSpec): Promise<{ taskId: string }>; + getTask(taskId: string): Promise; claimTask(): Promise; completeTask(options: { taskId: string;