From b45a34fb15e50593470afc73c14647d4d78a9ed2 Mon Sep 17 00:00:00 2001 From: Dominik Henneke Date: Mon, 25 Oct 2021 11:27:28 +0200 Subject: [PATCH] Adds a new endpoint for consuming logs from the Scaffolder that uses long polling instead of Server Sent Events Signed-off-by: Dominik Henneke --- .changeset/selfish-countries-prove.md | 42 ++++ .../src/service/router.test.ts | 80 ++++++++ .../scaffolder-backend/src/service/router.ts | 41 +++- plugins/scaffolder/api-report.md | 9 +- plugins/scaffolder/package.json | 1 + plugins/scaffolder/src/api.test.ts | 179 ++++++++++++++++++ plugins/scaffolder/src/api.ts | 66 ++++++- plugins/scaffolder/src/setupTests.ts | 1 + 8 files changed, 404 insertions(+), 15 deletions(-) create mode 100644 .changeset/selfish-countries-prove.md diff --git a/.changeset/selfish-countries-prove.md b/.changeset/selfish-countries-prove.md new file mode 100644 index 0000000000..03e36f6d80 --- /dev/null +++ b/.changeset/selfish-countries-prove.md @@ -0,0 +1,42 @@ +--- +'@backstage/plugin-scaffolder': patch +'@backstage/plugin-scaffolder-backend': patch +--- + +Adds a new endpoint for consuming logs from the Scaffolder that uses long polling instead of Server Sent Events. + +This is useful if Backstage is accessed from an environment that doesn't support SSE correctly, which happens in combination with certain enterprise HTTP Proxy servers. + +It is intended to switch the endpoint globally for the whole instance. +If you want to use it, you can provide a reconfigured API to the `scaffolderApiRef`: + +```tsx +// packages/app/src/apis.ts + +// ... +import { + scaffolderApiRef, + ScaffolderClient, +} from '@backstage/plugin-scaffolder'; + +export const apis: AnyApiFactory[] = [ + // ... + + createApiFactory({ + api: scaffolderApiRef, + deps: { + discoveryApi: discoveryApiRef, + identityApi: identityApiRef, + scmIntegrationsApi: scmIntegrationsApiRef, + }, + factory: ({ discoveryApi, identityApi, scmIntegrationsApi }) => + new ScaffolderClient({ + discoveryApi, + identityApi, + scmIntegrationsApi, + // use long polling instead of an eventsource + useLongPollingLogs: true, + }), + }), +]; +``` diff --git a/plugins/scaffolder-backend/src/service/router.test.ts b/plugins/scaffolder-backend/src/service/router.test.ts index 5157af4eaf..d6136fe872 100644 --- a/plugins/scaffolder-backend/src/service/router.test.ts +++ b/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); + }); + }); }); diff --git a/plugins/scaffolder-backend/src/service/router.ts b/plugins/scaffolder-backend/src/service/router.ts index fce3b885c6..076a7b0036 100644 --- a/plugins/scaffolder-backend/src/service/router.ts +++ b/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(); diff --git a/plugins/scaffolder/api-report.md b/plugins/scaffolder/api-report.md index 554fccabb7..0ac6910264 100644 --- a/plugins/scaffolder/api-report.md +++ b/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): Promise; // (undocumented) - streamLogs({ - taskId, - after, - }: { - taskId: string; - after?: number; - }): Observable; + streamLogs(opts: { taskId: string; after?: number }): Observable; } // Warning: (ae-missing-release-tag) "ScaffolderFieldExtensions" is exported by the package, but it is missing a release tag (@alpha, @beta, @public, or @internal) diff --git a/plugins/scaffolder/package.json b/plugins/scaffolder/package.json index 20b8e934f7..58bee32201 100644 --- a/plugins/scaffolder/package.json +++ b/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", diff --git a/plugins/scaffolder/src/api.test.ts b/plugins/scaffolder/src/api.test.ts index f11b2cf06c..8b53029cd7 100644 --- a/plugins/scaffolder/src/api.test.ts +++ b/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(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(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(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!' }, + }); + }); + }); }); }); diff --git a/plugins/scaffolder/src/api.ts b/plugins/scaffolder/src/api.ts index 1b531b652d..5f2874d653 100644 --- a/plugins/scaffolder/src/api.ts +++ b/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({ 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 { + 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 { + 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. */ diff --git a/plugins/scaffolder/src/setupTests.ts b/plugins/scaffolder/src/setupTests.ts index 84589060ec..10252413a7 100644 --- a/plugins/scaffolder/src/setupTests.ts +++ b/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;