Adds a new endpoint for consuming logs from the Scaffolder that uses long polling instead of Server Sent Events

Signed-off-by: Dominik Henneke <dominik.henneke@sda-se.com>

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

@@ -318,4 +318,84 @@ data: {"id":1,"taskId":"a-random-id","type":"completion","createdAt":"","body":{
       expect(unsubscribe).toBeCalledTimes(1);
     });
   });
+
+  describe('GET /v2/tasks/:taskId/logs', () => {
+    it('should return log messages', async () => {
+      const unsubscribe = jest.fn();
+      MockStorageTaskBroker.prototype.observe.mockImplementation(
+        ({ taskId }, callback) => {
+          callback(undefined, {
+            events: [
+              {
+                id: 0,
+                taskId,
+                type: 'log',
+                createdAt: '',
+                body: { message: 'My log message' },
+              },
+              {
+                id: 1,
+                taskId,
+                type: 'completion',
+                createdAt: '',
+                body: { message: 'Finished!' },
+              },
+            ],
+          });
+          return unsubscribe;
+        },
+      );
+
+      const response = await request(app).get('/v2/tasks/a-random-id/logs');
+
+      expect(response.status).toEqual(200);
+      expect(response.body).toEqual([
+        {
+          id: 0,
+          taskId: 'a-random-id',
+          type: 'log',
+          createdAt: '',
+          body: { message: 'My log message' },
+        },
+        {
+          id: 1,
+          taskId: 'a-random-id',
+          type: 'completion',
+          createdAt: '',
+          body: { message: 'Finished!' },
+        },
+      ]);
+
+      expect(MockStorageTaskBroker.prototype.observe).toBeCalledTimes(1);
+      expect(MockStorageTaskBroker.prototype.observe).toBeCalledWith(
+        { taskId: 'a-random-id' },
+        expect.any(Function),
+      );
+      expect(unsubscribe).toBeCalledTimes(1);
+    });
+
+    it('should return log messages with after query', async () => {
+      const unsubscribe = jest.fn();
+      MockStorageTaskBroker.prototype.observe.mockImplementation(
+        (_, callback) => {
+          callback(undefined, { events: [] });
+          return unsubscribe;
+        },
+      );
+
+      const response = await request(app)
+        .get('/v2/tasks/a-random-id/logs')
+        .query({ after: 10 });
+
+      expect(response.status).toEqual(200);
+      expect(response.body).toEqual([]);
+
+      expect(MockStorageTaskBroker.prototype.observe).toBeCalledTimes(1);
+      expect(MockStorageTaskBroker.prototype.observe).toBeCalledWith(
+        { taskId: 'a-random-id', after: 10 },
+        expect.any(Function),
+      );
+      expect(unsubscribe).toBeCalledTimes(1);
+    });
+  });
 });

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

@@ -247,7 +247,9 @@ export async function createRouter(
     })
     .get('/v2/tasks/:taskId/eventstream', async (req, res) => {
       const { taskId } = req.params;
-      const after = Number(req.query.after) || undefined;
+      const after =
+        req.query.after !== undefined ? Number(req.query.after) : undefined;
+
       logger.debug(`Event stream observing taskId '${taskId}' opened`);
 
       // Mandatory headers and http status to keep connection open
@@ -289,6 +291,43 @@ export async function createRouter(
         unsubscribe();
         logger.debug(`Event stream observing taskId '${taskId}' closed`);
       });
+    })
+    .get('/v2/tasks/:taskId/logs', async (req, res) => {
+      const { taskId } = req.params;
+      const after = Number(req.query.after) || undefined;
+
+      let unsubscribe = () => {};
+
+      // cancel the request after 30 seconds. this aligns with the recommendations of RFC 6202.
+      const timeout = setTimeout(() => {
+        unsubscribe();
+        res.json([]);
+      }, 30_000);
+
+      // Get all known events after an id (always includes the completion event) and return the first callback
+      unsubscribe = taskBroker.observe(
+        { taskId, after },
+        (error, { events }) => {
+          // stop the timeout
+          clearTimeout(timeout);
+          unsubscribe();
+
+          if (error) {
+            logger.error(
+              `Received error from log when observing taskId '${taskId}', ${error}`,
+            );
+          }
+
+          res.json(events);
+        },
+      );
+
+      // When client closes connection we update the clients list
+      // avoiding the disconnected one
+      req.on('close', () => {
+        unsubscribe();
+        clearTimeout(timeout);
+      });
     });
 
   const app = express();

plugins/scaffolder/api-report.md

@@ -178,6 +178,7 @@ export class ScaffolderClient implements ScaffolderApi {
     discoveryApi: DiscoveryApi;
     identityApi: IdentityApi;
     scmIntegrationsApi: ScmIntegrationRegistry;
+    useLongPollingLogs?: boolean;
   });
   // (undocumented)
   getIntegrationsList(options: { allowedHosts: string[] }): Promise<
@@ -199,13 +200,7 @@ export class ScaffolderClient implements ScaffolderApi {
   // Warning: (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen
   scaffold(templateName: string, values: Record<string, any>): Promise<string>;
   // (undocumented)
-  streamLogs({
-    taskId,
-    after,
-  }: {
-    taskId: string;
-    after?: number;
-  }): Observable<LogEvent>;
+  streamLogs(opts: { taskId: string; after?: number }): Observable<LogEvent>;
 }
 
 // Warning: (ae-missing-release-tag) "ScaffolderFieldExtensions" is exported by the package, but it is missing a release tag (@alpha, @beta, @public, or @internal)

plugins/scaffolder/package.json

@@ -55,6 +55,7 @@
     "json-schema": "^0.3.0",
     "lodash": "^4.17.21",
     "luxon": "^2.0.2",
+    "qs": "^6.9.4",
     "react": "^16.13.1",
     "react-dom": "^16.13.1",
     "react-lazylog": "^4.5.2",

plugins/scaffolder/src/api.test.ts

@@ -16,13 +16,19 @@
 
 import { ConfigReader } from '@backstage/core-app-api';
 import { ScmIntegrations } from '@backstage/integration';
+import { setupRequestMockHandlers } from '@backstage/test-utils';
+import { rest } from 'msw';
+import { setupServer } from 'msw/node';
 import { ScaffolderClient } from './api';
 
 const MockedEventSource = global.EventSource as jest.MockedClass<
   typeof EventSource
 >;
 
+const server = setupServer();
+
 describe('api', () => {
+  setupRequestMockHandlers(server);
   const mockBaseUrl = 'http://backstage/api';
 
   const discoveryApi = { getBaseUrl: async () => mockBaseUrl };
@@ -114,5 +120,178 @@ describe('api', () => {
         });
       });
     });
+
+    describe('longPolling', () => {
+      beforeEach(() => {
+        apiClient = new ScaffolderClient({
+          scmIntegrationsApi,
+          discoveryApi,
+          identityApi,
+          useLongPollingLogs: true,
+        });
+      });
+
+      it('should work', async () => {
+        server.use(
+          rest.get(`${mockBaseUrl}/v2/tasks/:taskId/logs`, (req, res, ctx) => {
+            const { taskId } = req.params;
+            const after = req.url.searchParams.get('after');
+
+            if (taskId === 'a-random-task-id') {
+              if (!after) {
+                return res(
+                  ctx.json([
+                    {
+                      id: 1,
+                      taskId: 'a-random-id',
+                      type: 'log',
+                      createdAt: '',
+                      body: { message: 'My log message' },
+                    },
+                  ]),
+                );
+              } else if (after === '1') {
+                return res(
+                  ctx.json([
+                    {
+                      id: 2,
+                      taskId: 'a-random-id',
+                      type: 'completion',
+                      createdAt: '',
+                      body: { message: 'Finished!' },
+                    },
+                  ]),
+                );
+              }
+            }
+
+            return res(ctx.status(500));
+          }),
+        );
+
+        const next = jest.fn();
+
+        await new Promise<void>(complete =>
+          apiClient
+            .streamLogs({ taskId: 'a-random-task-id' })
+            .subscribe({ next, complete }),
+        );
+
+        expect(next).toBeCalledTimes(2);
+        expect(next).toBeCalledWith({
+          id: 1,
+          taskId: 'a-random-id',
+          type: 'log',
+          createdAt: '',
+          body: { message: 'My log message' },
+        });
+        expect(next).toBeCalledWith({
+          id: 2,
+          taskId: 'a-random-id',
+          type: 'completion',
+          createdAt: '',
+          body: { message: 'Finished!' },
+        });
+      });
+
+      it('should unsubscribe', async () => {
+        expect.assertions(3);
+
+        server.use(
+          rest.get(`${mockBaseUrl}/v2/tasks/:taskId/logs`, (req, res, ctx) => {
+            const { taskId } = req.params;
+
+            const after = req.url.searchParams.get('after');
+
+            // use assertion to make sure it is not called after unsubscribing
+            expect(after).toBe(null);
+
+            if (taskId === 'a-random-task-id') {
+              return res(
+                ctx.json([
+                  {
+                    id: 1,
+                    taskId: 'a-random-id',
+                    type: 'log',
+                    createdAt: '',
+                    body: { message: 'My log message' },
+                  },
+                ]),
+              );
+            }
+
+            return res(ctx.status(500));
+          }),
+        );
+
+        const next = jest.fn();
+
+        await new Promise<void>(complete => {
+          const subscription = apiClient
+            .streamLogs({ taskId: 'a-random-task-id' })
+            .subscribe({
+              next: (...args) => {
+                next(...args);
+                subscription.unsubscribe();
+                complete();
+              },
+            });
+        });
+
+        expect(next).toBeCalledTimes(1);
+        expect(next).toBeCalledWith({
+          id: 1,
+          taskId: 'a-random-id',
+          type: 'log',
+          createdAt: '',
+          body: { message: 'My log message' },
+        });
+      });
+
+      it('should continue after error', async () => {
+        const called = jest.fn();
+
+        server.use(
+          rest.get(`${mockBaseUrl}/v2/tasks/:taskId/logs`, (_req, res, ctx) => {
+            called();
+
+            if (called.mock.calls.length > 1) {
+              return res(
+                ctx.json([
+                  {
+                    id: 2,
+                    taskId: 'a-random-id',
+                    type: 'completion',
+                    createdAt: '',
+                    body: { message: 'Finished!' },
+                  },
+                ]),
+              );
+            }
+
+            return res(ctx.status(500));
+          }),
+        );
+
+        const next = jest.fn();
+
+        await new Promise<void>(complete =>
+          apiClient
+            .streamLogs({ taskId: 'a-random-task-id' })
+            .subscribe({ next, complete }),
+        );
+
+        expect(called).toBeCalledTimes(2);
+
+        expect(next).toBeCalledTimes(1);
+        expect(next).toBeCalledWith({
+          id: 2,
+          taskId: 'a-random-id',
+          type: 'completion',
+          createdAt: '',
+          body: { message: 'Finished!' },
+        });
+      });
+    });
   });
 });

plugins/scaffolder/src/api.ts

@@ -15,17 +15,18 @@
  */
 
 import { EntityName } from '@backstage/catalog-model';
-import { JsonObject, JsonValue, Observable } from '@backstage/types';
-import { ResponseError } from '@backstage/errors';
-import { ScmIntegrationRegistry } from '@backstage/integration';
-import { Field, FieldValidation } from '@rjsf/core';
-import ObservableImpl from 'zen-observable';
-import { ListActionsResponse, ScaffolderTask, Status } from './types';
 import {
   createApiRef,
   DiscoveryApi,
   IdentityApi,
 } from '@backstage/core-plugin-api';
+import { ResponseError } from '@backstage/errors';
+import { ScmIntegrationRegistry } from '@backstage/integration';
+import { JsonObject, JsonValue, Observable } from '@backstage/types';
+import { Field, FieldValidation } from '@rjsf/core';
+import qs from 'qs';
+import ObservableImpl from 'zen-observable';
+import { ListActionsResponse, ScaffolderTask, Status } from './types';
 
 export const scaffolderApiRef = createApiRef<ScaffolderApi>({
   id: 'plugin.scaffolder.service',
@@ -94,15 +95,18 @@ export class ScaffolderClient implements ScaffolderApi {
   private readonly discoveryApi: DiscoveryApi;
   private readonly identityApi: IdentityApi;
   private readonly scmIntegrationsApi: ScmIntegrationRegistry;
+  private readonly useLongPollingLogs: boolean;
 
   constructor(options: {
     discoveryApi: DiscoveryApi;
     identityApi: IdentityApi;
     scmIntegrationsApi: ScmIntegrationRegistry;
+    useLongPollingLogs?: boolean;
   }) {
     this.discoveryApi = options.discoveryApi;
     this.identityApi = options.identityApi;
     this.scmIntegrationsApi = options.scmIntegrationsApi;
+    this.useLongPollingLogs = options.useLongPollingLogs ?? false;
   }
 
   async getIntegrationsList(options: { allowedHosts: string[] }) {
@@ -189,7 +193,15 @@ export class ScaffolderClient implements ScaffolderApi {
     return await response.json();
   }
 
-  streamLogs({
+  streamLogs(opts: { taskId: string; after?: number }): Observable<LogEvent> {
+    if (this.useLongPollingLogs) {
+      return this.streamLogsPolling(opts);
+    }
+
+    return this.streamLogsEventStream(opts);
+  }
+
+  private streamLogsEventStream({
     taskId,
     after,
   }: {
@@ -239,6 +251,46 @@ export class ScaffolderClient implements ScaffolderApi {
     });
   }
 
+  private streamLogsPolling({
+    taskId,
+    after: inputAfter,
+  }: {
+    taskId: string;
+    after?: number;
+  }): Observable<LogEvent> {
+    let after = inputAfter;
+
+    return new ObservableImpl(subscriber => {
+      this.discoveryApi.getBaseUrl('scaffolder').then(async baseUrl => {
+        while (!subscriber.closed) {
+          const url = `${baseUrl}/v2/tasks/${encodeURIComponent(
+            taskId,
+          )}/logs?${qs.stringify({ after })}`;
+          const response = await fetch(url);
+
+          if (!response.ok) {
+            // wait for one second to not run into an
+            await new Promise(resolve => setTimeout(resolve, 1000));
+            continue;
+          }
+
+          const logs = (await response.json()) as LogEvent[];
+
+          for (const event of logs) {
+            after = Number(event.id);
+
+            subscriber.next(event);
+
+            if (event.type === 'completion') {
+              subscriber.complete();
+              return;
+            }
+          }
+        }
+      });
+    });
+  }
+
   /**
    * @returns ListActionsResponse containing all registered actions.
    */

plugins/scaffolder/src/setupTests.ts

@@ -15,6 +15,7 @@
  */
 
 import '@testing-library/jest-dom';
+import 'cross-fetch/polyfill';
 
 const { EventSourcePolyfill } = jest.requireMock('event-source-polyfill');
 global.EventSource = EventSourcePolyfill;