Merge pull request #7758 from SDA-SE/feat/scaffolder-long-polling

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

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

@@ -29,24 +29,25 @@ jest.doMock('fs-extra', () => ({
 }));
 
 import {
+  DatabaseManager,
+  DockerContainerRunner,
   getVoidLogger,
   PluginDatabaseManager,
-  DatabaseManager,
   UrlReaders,
-  DockerContainerRunner,
 } from '@backstage/backend-common';
 import { CatalogApi } from '@backstage/catalog-client';
+import { TemplateEntityV1beta2 } from '@backstage/catalog-model';
 import { ConfigReader } from '@backstage/config';
 import express from 'express';
 import request from 'supertest';
-import { TemplateEntityV1beta2 } from '@backstage/catalog-model';
 /**
  * TODO: The following should import directly from the router file.
  * Due to a circular dependency between this plugin and the
  * plugin-scaffolder-backend-module-cookiecutter plugin, it results in an error:
  * TypeError: _pluginscaffolderbackend.createTemplateAction is not a function
  */
-import { createRouter } from '../index';
+import { createRouter, DatabaseTaskStore, TaskBroker } from '../index';
+import { StorageTaskBroker } from '../scaffolder/tasks/StorageTaskBroker';
 
 const createCatalogClient = (templates: any[] = []) =>
   ({
@@ -73,6 +74,7 @@ const mockUrlReader = UrlReaders.default({
 
 describe('createRouter', () => {
   let app: express.Express;
+  let taskBroker: TaskBroker;
   const template: TemplateEntityV1beta2 = {
     apiVersion: 'backstage.io/v1beta2',
     kind: 'Template',
@@ -102,7 +104,17 @@ describe('createRouter', () => {
     },
   };
 
-  beforeAll(async () => {
+  beforeEach(async () => {
+    const logger = getVoidLogger();
+    const databaseTaskStore = await DatabaseTaskStore.create({
+      database: await createDatabase().getClient(),
+    });
+    taskBroker = new StorageTaskBroker(databaseTaskStore, logger);
+
+    jest.spyOn(taskBroker, 'dispatch');
+    jest.spyOn(taskBroker, 'get');
+    jest.spyOn(taskBroker, 'observe');
+
     const router = await createRouter({
       logger: getVoidLogger(),
       config: new ConfigReader({}),
@@ -110,11 +122,12 @@ describe('createRouter', () => {
       catalogClient: createCatalogClient([template]),
       containerRunner: new DockerContainerRunner({} as any),
       reader: mockUrlReader,
+      taskBroker,
     });
     app = express().use(router);
   });
 
-  beforeEach(() => {
+  afterEach(() => {
     jest.resetAllMocks();
   });
 
@@ -142,6 +155,12 @@ describe('createRouter', () => {
     });
 
     it('return the template id', async () => {
+      (
+        taskBroker.dispatch as jest.Mocked<TaskBroker>['dispatch']
+      ).mockResolvedValue({
+        taskId: 'a-random-id',
+      });
+
       const response = await request(app)
         .post('/v2/tasks')
         .send({
@@ -151,28 +170,241 @@ describe('createRouter', () => {
           },
         });
 
-      expect(response.body.id).toBeDefined();
+      expect(response.body.id).toBe('a-random-id');
       expect(response.status).toEqual(201);
     });
   });
 
   describe('GET /v2/tasks/:taskId', () => {
     it('does not divulge secrets', async () => {
-      const postResponse = await request(app)
-        .post('/v2/tasks')
-        .set('Authorization', 'Bearer secret')
-        .send({
-          templateName: 'create-react-app-template',
-          values: {
-            required: 'required-value',
-          },
-        });
+      (taskBroker.get as jest.Mocked<TaskBroker>['get']).mockResolvedValue({
+        id: 'a-random-id',
+        spec: {} as any,
+        status: 'completed',
+        createdAt: '',
+        secrets: { token: 'secret' },
+      });
 
-      const response = await request(app)
-        .get(`/v2/tasks/${postResponse.body.id}`)
-        .send();
+      const response = await request(app).get(`/v2/tasks/a-random-id`);
       expect(response.status).toEqual(200);
+      expect(response.body.status).toBe('completed');
       expect(response.body.secrets).toBeUndefined();
     });
   });
+
+  describe('GET /v2/tasks/:taskId/eventstream', () => {
+    it('should return log messages', async () => {
+      const unsubscribe = jest.fn();
+      (
+        taskBroker.observe as jest.Mocked<TaskBroker>['observe']
+      ).mockImplementation(({ taskId }, callback) => {
+        // emit after this function returned
+        setImmediate(() => {
+          callback(undefined, {
+            events: [
+              {
+                id: 0,
+                taskId,
+                type: 'log',
+                createdAt: '',
+                body: { message: 'My log message' },
+              },
+            ],
+          });
+          callback(undefined, {
+            events: [
+              {
+                id: 1,
+                taskId,
+                type: 'completion',
+                createdAt: '',
+                body: { message: 'Finished!' },
+              },
+            ],
+          });
+        });
+
+        return { unsubscribe };
+      });
+
+      let statusCode: any = undefined;
+      let headers: any = {};
+      const responseDataFn = jest.fn();
+
+      const req = request(app)
+        .get('/v2/tasks/a-random-id/eventstream')
+        .set('accept', 'text/event-stream')
+        .parse((res, _) => {
+          ({ statusCode, headers } = res as any);
+
+          res.on('data', chunk => {
+            responseDataFn(chunk.toString());
+
+            // the server expects the client to abort the request
+            if (chunk.includes('completion')) {
+              req.abort();
+            }
+          });
+        });
+
+      // wait for the request to finish
+      await req.catch(() => {
+        // ignore 'aborted' error
+      });
+
+      expect(statusCode).toBe(200);
+      expect(headers['content-type']).toBe('text/event-stream');
+      expect(responseDataFn).toBeCalledTimes(2);
+      expect(responseDataFn).toBeCalledWith(`event: log
+data: {"id":0,"taskId":"a-random-id","type":"log","createdAt":"","body":{"message":"My log message"}}
+
+`);
+      expect(responseDataFn).toBeCalledWith(`event: completion
+data: {"id":1,"taskId":"a-random-id","type":"completion","createdAt":"","body":{"message":"Finished!"}}
+
+`);
+
+      expect(taskBroker.observe).toBeCalledTimes(1);
+      expect(taskBroker.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();
+      (
+        taskBroker.observe as jest.Mocked<TaskBroker>['observe']
+      ).mockImplementation(({ taskId }, callback) => {
+        setImmediate(() => {
+          callback(undefined, {
+            events: [
+              {
+                id: 1,
+                taskId,
+                type: 'completion',
+                createdAt: '',
+                body: { message: 'Finished!' },
+              },
+            ],
+          });
+        });
+        return { unsubscribe };
+      });
+
+      let statusCode: any = undefined;
+      let headers: any = {};
+
+      const req = request(app)
+        .get('/v2/tasks/a-random-id/eventstream')
+        .query({ after: 10 })
+        .set('accept', 'text/event-stream')
+        .parse((res, _) => {
+          ({ statusCode, headers } = res as any);
+
+          res.on('data', () => {
+            // close immediately
+            req.abort();
+          });
+        });
+
+      // wait for the request to finish
+      await req.catch(() => {
+        // ignore 'aborted' error
+      });
+
+      expect(statusCode).toBe(200);
+      expect(headers['content-type']).toBe('text/event-stream');
+
+      expect(taskBroker.observe).toBeCalledTimes(1);
+      expect(taskBroker.observe).toBeCalledWith(
+        { taskId: 'a-random-id', after: 10 },
+        expect.any(Function),
+      );
+
+      expect(unsubscribe).toBeCalledTimes(1);
+    });
+  });
+
+  describe('GET /v2/tasks/:taskId/events', () => {
+    it('should return log messages', async () => {
+      const unsubscribe = jest.fn();
+      (
+        taskBroker.observe as jest.Mocked<TaskBroker>['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/events');
+
+      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(taskBroker.observe).toBeCalledTimes(1);
+      expect(taskBroker.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();
+      (
+        taskBroker.observe as jest.Mocked<TaskBroker>['observe']
+      ).mockImplementation((_, callback) => {
+        callback(undefined, { events: [] });
+        return { unsubscribe };
+      });
+
+      const response = await request(app)
+        .get('/v2/tasks/a-random-id/events')
+        .query({ after: 10 });
+
+      expect(response.status).toEqual(200);
+      expect(response.body).toEqual([]);
+
+      expect(taskBroker.observe).toBeCalledTimes(1);
+      expect(taskBroker.observe).toBeCalledWith(
+        { taskId: 'a-random-id', after: 10 },
+        expect.any(Function),
+      );
+      expect(unsubscribe).toBeCalledTimes(1);
+    });
+  });
 });

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

@@ -14,33 +14,33 @@
  * limitations under the License.
  */
 
-import { Config } from '@backstage/config';
-import express from 'express';
-import Router from 'express-promise-router';
-import { Logger } from 'winston';
-import { CatalogEntityClient } from '../lib/catalog';
-import { validate } from 'jsonschema';
-import {
-  DatabaseTaskStore,
-  TemplateActionRegistry,
-  TaskWorker,
-  TemplateAction,
-  createBuiltinActions,
-} from '../scaffolder';
-import { StorageTaskBroker } from '../scaffolder/tasks/StorageTaskBroker';
-import { getEntityBaseUrl, getWorkingDirectory } from './helpers';
 import {
   ContainerRunner,
   PluginDatabaseManager,
   UrlReader,
 } from '@backstage/backend-common';
-import { InputError, NotFoundError } from '@backstage/errors';
 import { CatalogApi } from '@backstage/catalog-client';
-import { TemplateEntityV1beta2, Entity } from '@backstage/catalog-model';
-import { TemplateEntityV1beta3 } from '@backstage/plugin-scaffolder-common';
-
+import { Entity, TemplateEntityV1beta2 } from '@backstage/catalog-model';
+import { Config } from '@backstage/config';
+import { InputError, NotFoundError } from '@backstage/errors';
 import { ScmIntegrations } from '@backstage/integration';
-import { TaskBroker, TaskSpec } from '../scaffolder/tasks/types';
+import { TemplateEntityV1beta3 } from '@backstage/plugin-scaffolder-common';
+import express from 'express';
+import Router from 'express-promise-router';
+import { validate } from 'jsonschema';
+import { Logger } from 'winston';
+import { CatalogEntityClient } from '../lib/catalog';
+import {
+  createBuiltinActions,
+  DatabaseTaskStore,
+  TaskBroker,
+  TaskSpec,
+  TaskWorker,
+  TemplateAction,
+  TemplateActionRegistry,
+} from '../scaffolder';
+import { StorageTaskBroker } from '../scaffolder/tasks/StorageTaskBroker';
+import { getEntityBaseUrl, getWorkingDirectory } from './helpers';
 
 /**
  * RouterOptions
@@ -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
@@ -278,7 +280,8 @@ export async function createRouter(
               // to automatically reconnect because it lost connection.
             }
           }
-          res.flush();
+          // res.flush() is only available with the compression middleware
+          res.flush?.();
           if (shouldUnsubscribe) unsubscribe();
         },
       );
@@ -288,6 +291,43 @@ export async function createRouter(
         unsubscribe();
         logger.debug(`Event stream observing taskId '${taskId}' closed`);
       });
+    })
+    .get('/v2/tasks/:taskId/events', 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",
@@ -77,6 +78,7 @@
     "@types/jest": "^26.0.7",
     "@types/node": "^14.14.32",
     "cross-fetch": "^3.0.6",
+    "event-source-polyfill": "^1.0.25",
     "msw": "^0.35.0"
   },
   "files": [

plugins/scaffolder/src/api.test.ts

@@ -14,12 +14,24 @@
  * limitations under the License.
  */
 
-import { ScmIntegrations } from '@backstage/integration';
-import { ScaffolderClient } from './api';
 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', () => {
-  const discoveryApi = {} as any;
+  setupRequestMockHandlers(server);
+  const mockBaseUrl = 'http://backstage/api';
+
+  const discoveryApi = { getBaseUrl: async () => mockBaseUrl };
   const identityApi = {} as any;
   const scmIntegrationsApi = ScmIntegrations.fromConfig(
     new ConfigReader({
@@ -32,10 +44,14 @@ describe('api', () => {
       },
     }),
   );
-  const apiClient = new ScaffolderClient({
-    scmIntegrationsApi,
-    discoveryApi,
-    identityApi,
+
+  let apiClient: ScaffolderClient;
+  beforeEach(() => {
+    apiClient = new ScaffolderClient({
+      scmIntegrationsApi,
+      discoveryApi,
+      identityApi,
+    });
   });
 
   it('should return default and custom integrations', async () => {
@@ -51,4 +67,240 @@ describe('api', () => {
       expect(allowedHosts).toContain(integration.host),
     );
   });
+
+  describe('streamEvents', () => {
+    describe('eventsource', () => {
+      it('should work', async () => {
+        MockedEventSource.prototype.addEventListener.mockImplementation(
+          (type, fn) => {
+            if (typeof fn !== 'function') {
+              return;
+            }
+
+            if (type === 'log') {
+              fn({
+                data: '{"id":1,"taskId":"a-random-id","type":"log","createdAt":"","body":{"message":"My log message"}}',
+              } as any);
+            } else if (type === 'completion') {
+              fn({
+                data: '{"id":2,"taskId":"a-random-id","type":"completion","createdAt":"","body":{"message":"Finished!"}}',
+              } as any);
+            }
+          },
+        );
+
+        const next = jest.fn();
+
+        await new Promise<void>(complete => {
+          apiClient
+            .streamLogs({ taskId: 'a-random-task-id' })
+            .subscribe({ next, complete });
+        });
+
+        expect(MockedEventSource).toBeCalledWith(
+          'http://backstage/api/v2/tasks/a-random-task-id/eventstream',
+          { withCredentials: true },
+        );
+        expect(MockedEventSource.prototype.close).toBeCalled();
+
+        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!' },
+        });
+      });
+    });
+
+    describe('longPolling', () => {
+      beforeEach(() => {
+        apiClient = new ScaffolderClient({
+          scmIntegrationsApi,
+          discoveryApi,
+          identityApi,
+          useLongPollingLogs: true,
+        });
+      });
+
+      it('should work', async () => {
+        server.use(
+          rest.get(
+            `${mockBaseUrl}/v2/tasks/:taskId/events`,
+            (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/events`,
+            (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/events`,
+            (_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,
+          )}/events?${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,3 +15,7 @@
  */
 
 import '@testing-library/jest-dom';
+import 'cross-fetch/polyfill';
+
+const { EventSourcePolyfill } = jest.requireMock('event-source-polyfill');
+global.EventSource = EventSourcePolyfill;