Add log metadata and getTask method

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<DbTaskRow> {
+  async getTask(taskId: string): Promise<DbTaskRow> {
     const [result] = await this.db<RawDbTaskRow>('tasks')
       .where({ id: taskId })
       .select();

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

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<void> {
+  async emitLog(message: string, metadata?: JsonObject): Promise<void> {
     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<DbTaskRow> {
+    return this.storage.getTask(taskId);
+  }
+
   observe(
     options: {
       taskId: string;

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

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<void>;
+  emitLog(message: string, metadata?: JsonValue): Promise<void>;
   complete(result: CompletedTaskState): Promise<void>;
   getWorkspaceName(): Promise<string>;
 }
@@ -90,6 +90,7 @@ export type TaskStoreGetEventsOptions = {
 };
 export interface TaskStore {
   createTask(task: TaskSpec): Promise<{ taskId: string }>;
+  getTask(taskId: string): Promise<DbTaskRow>;
   claimTask(): Promise<DbTaskRow | undefined>;
   completeTask(options: {
     taskId: string;