From ea33c1240a065fe5c865710480c50051db8d2eaa Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Fri, 24 May 2024 17:10:57 +0200 Subject: [PATCH] events-node: update DefaultEventsService to use backend Signed-off-by: Patrik Oldsberg --- plugins/events-node/package.json | 1 + .../src/api/DefaultEventsService.ts | 359 +++++++++++++++--- plugins/events-node/src/service.ts | 11 +- 3 files changed, 307 insertions(+), 64 deletions(-) diff --git a/plugins/events-node/package.json b/plugins/events-node/package.json index de880dc399..2ac1a6922b 100644 --- a/plugins/events-node/package.json +++ b/plugins/events-node/package.json @@ -52,6 +52,7 @@ }, "dependencies": { "@backstage/backend-plugin-api": "workspace:^", + "@backstage/errors": "workspace:^", "@backstage/types": "workspace:^", "cross-fetch": "^4.0.0", "uri-template": "^2.0.0" diff --git a/plugins/events-node/src/api/DefaultEventsService.ts b/plugins/events-node/src/api/DefaultEventsService.ts index bb5c2a0ca8..d1d26b1723 100644 --- a/plugins/events-node/src/api/DefaultEventsService.ts +++ b/plugins/events-node/src/api/DefaultEventsService.ts @@ -14,9 +14,267 @@ * limitations under the License. */ -import { LoggerService } from '@backstage/backend-plugin-api'; +import { + AuthService, + DiscoveryService, + LoggerService, +} from '@backstage/backend-plugin-api'; import { EventParams } from './EventParams'; import { EventsService, EventsServiceSubscribeOptions } from './EventsService'; +import { DefaultApiClient } from '../generated'; +import { NotImplementedError, ResponseError } from '@backstage/errors'; + +const POLL_BACKOFF_START_MS = 1_000; +const POLL_BACKOFF_MAX_MS = 60_000; +const POLL_BACKOFF_FACTOR = 2; + +/** + * Local event bus for subscribers within the same process. + * + * When publishing events we'll keep track of which subscribers we managed to + * reach locally, and forward those subscriber IDs to the events backend if it + * is in use. The events backend will then both avoid forwarding the same events + * to those subscribers again, but also avoid storing the event altogether if + * there are no other subscribers. + * @internal + */ +export class LocalEventBus { + readonly #logger: LoggerService; + + readonly #subscribers = new Map< + string, + Omit[] + >(); + + constructor(logger: LoggerService) { + this.#logger = logger; + } + + async publish( + params: EventParams, + ): Promise<{ notifiedSubscribers: string[] }> { + this.#logger.debug( + `Event received: topic=${params.topic}, metadata=${JSON.stringify( + params.metadata, + )}, payload=${JSON.stringify(params.eventPayload)}`, + ); + + if (!this.#subscribers.has(params.topic)) { + return { notifiedSubscribers: [] }; + } + + const onEventPromises: Promise[] = []; + this.#subscribers.get(params.topic)?.forEach(subscription => { + onEventPromises.push( + (async () => { + try { + await subscription.onEvent(params); + } catch (error) { + this.#logger.warn( + `Subscriber "${subscription.id}" failed to process event for topic "${params.topic}"`, + error, + ); + } + return subscription.id; + })(), + ); + }); + + return { notifiedSubscribers: await Promise.all(onEventPromises) }; + } + + async subscribe(options: EventsServiceSubscribeOptions): Promise { + options.topics.forEach(topic => { + if (!this.#subscribers.has(topic)) { + this.#subscribers.set(topic, []); + } + + this.#subscribers.get(topic)!.push({ + id: options.id, + onEvent: options.onEvent, + }); + }); + } +} + +/** + * Plugin specific events bus that delegates to the local bus, as well as the + * events backend if it is available. + */ +class PluginEventsService implements EventsService { + constructor( + private readonly pluginId: string, + private readonly localBus: LocalEventBus, + private readonly logger: LoggerService, + private client?: DefaultApiClient, + private readonly auth?: AuthService, + ) {} + + async publish(params: EventParams): Promise { + const { notifiedSubscribers } = await this.localBus.publish(params); + + if (!this.client) { + return; + } + const token = await this.#getToken(); + if (!token) { + return; + } + const res = await this.client.postEvent( + { + body: { + event: { payload: params.eventPayload, topic: params.topic }, + subscriptionIds: notifiedSubscribers, + }, + }, + { token }, + ); + + if (!res.ok) { + if (res.status === 404) { + this.logger.warn( + `Event publish request failed with status 404, events backend not found. Future events will not be persisted.`, + ); + delete this.client; + return; + } + throw await ResponseError.fromResponse(res); + } + } + + async subscribe(options: EventsServiceSubscribeOptions): Promise { + const subscriptionId = `${this.pluginId}.${options.id}`; + + await this.localBus.subscribe({ + id: subscriptionId, + topics: options.topics, + onEvent: options.onEvent, + }); + + if (!this.client) { + return; + } + const token = await this.#getToken(); + if (!token) { + return; + } + const res = await this.client.putSubscription( + { + path: { subscriptionId }, + body: { topics: options.topics }, + }, + { token }, + ); + if (!res.ok) { + if (res.status === 404) { + this.logger.warn( + `Event subscribe request failed with status 404, events backend not found. Future events will not be persisted.`, + ); + delete this.client; + return; + } + throw await ResponseError.fromResponse(res); + } + + this.#startPolling(subscriptionId, options.onEvent); + } + + #startPolling( + subscriptionId: string, + onEvent: EventsServiceSubscribeOptions['onEvent'], + ) { + let backoffMs = POLL_BACKOFF_START_MS; + const poll = async () => { + if (!this.client) { + return; + } + try { + const token = await this.#getToken(); + if (!token) { + return; + } + const res = await this.client.getSubscriptionEvents( + { + path: { subscriptionId }, + }, + { token }, + ); + + if (!res.ok) { + throw await ResponseError.fromResponse(res); + } + backoffMs = POLL_BACKOFF_START_MS; + + // 204 arrives after a blocking response and means there are new events + // available or we timed out, either way should should try to read events + // immediately again + if (res.status === 204) { + process.nextTick(poll); + } else if (res.status === 200) { + const data = await res.json(); + if (data) { + for (const event of data.events ?? []) { + try { + await onEvent({ + topic: event.topic, + eventPayload: event.payload, + }); + } catch (error) { + this.logger.warn( + `Subscriber "${subscriptionId}" failed to process event for topic "${event.topic}"`, + error, + ); + } + } + } + process.nextTick(poll); + } else { + this.logger.warn( + `Unexpected response status ${res.status} from events backend for subscription "${subscriptionId}"`, + ); + } + } catch (error) { + this.logger.warn( + `Poll failed for subscription "${subscriptionId}", retrying in ${backoffMs.toFixed( + 0, + )}ms`, + error, + ); + setTimeout(poll, backoffMs); + backoffMs = Math.min( + backoffMs * POLL_BACKOFF_FACTOR, + POLL_BACKOFF_MAX_MS, + ); + } + }; + poll(); + } + + async #getToken() { + if (!this.auth) { + throw new Error('Auth service not available'); + } + + try { + const { token } = await this.auth.getPluginRequestToken({ + onBehalfOf: await this.auth.getOwnServiceCredentials(), + targetPluginId: 'events', + }); + return token; + } catch (error) { + // This is a bit hacky, but handles the case where new auth is used + // without legacy auth fallback, and the events backend is not installed + if (String(error).includes('Unable to generate legacy token')) { + this.logger.warn( + `The events backend is not available and neither is legacy auth. Future events will not be persisted.`, + ); + delete this.client; + return undefined; + } + throw error; + } + } +} /** * In-process event broker which will pass the event to all registered subscribers @@ -28,15 +286,16 @@ import { EventsService, EventsServiceSubscribeOptions } from './EventsService'; */ // TODO(pjungermann): add opentelemetry? (see plugins/catalog-backend/src/util/opentelemetry.ts, etc.) export class DefaultEventsService implements EventsService { - private readonly subscribers = new Map< - string, - Omit[] - >(); - - private constructor(private readonly logger: LoggerService) {} + private constructor( + private readonly logger: LoggerService, + private readonly localBus: LocalEventBus, + ) {} static create(options: { logger: LoggerService }): DefaultEventsService { - return new DefaultEventsService(options.logger); + return new DefaultEventsService( + options.logger, + new LocalEventBus(options.logger), + ); } /** @@ -45,60 +304,36 @@ export class DefaultEventsService implements EventsService { * * @param pluginId - The plugin that the `EventService` should be created for. */ - forPlugin(pluginId: string): EventsService { - return { - publish: (params: EventParams): Promise => { - return this.publish(params); - }, - subscribe: (options: EventsServiceSubscribeOptions): Promise => { - return this.subscribe({ - ...options, - id: `${pluginId}.${options.id}`, - }); - }, - }; - } - - async publish(params: EventParams): Promise { - this.logger.debug( - `Event received: topic=${params.topic}, metadata=${JSON.stringify( - params.metadata, - )}, payload=${JSON.stringify(params.eventPayload)}`, - ); - - if (!this.subscribers.has(params.topic)) { - return; - } - - const onEventPromises: Promise[] = []; - this.subscribers.get(params.topic)?.forEach(subscription => { - onEventPromises.push( - (async () => { - try { - await subscription.onEvent(params); - } catch (error) { - this.logger.warn( - `Subscriber "${subscription.id}" failed to process event for topic "${params.topic}"`, - error, - ); - } - })(), - ); - }); - - await Promise.all(onEventPromises); - } - - async subscribe(options: EventsServiceSubscribeOptions): Promise { - options.topics.forEach(topic => { - if (!this.subscribers.has(topic)) { - this.subscribers.set(topic, []); - } - - this.subscribers.get(topic)!.push({ - id: options.id, - onEvent: options.onEvent, + forPlugin( + pluginId: string, + options?: { + discovery: DiscoveryService; + logger: LoggerService; + auth: AuthService; + }, + ): EventsService { + const client = + options && + new DefaultApiClient({ + discoveryApi: options.discovery, }); - }); + const logger = options?.logger ?? this.logger; + return new PluginEventsService( + pluginId, + this.localBus, + logger, + client, + options?.auth, + ); + } + + /** @deprecated this method should not be called */ + async publish(_params: EventParams): Promise { + throw new NotImplementedError(); + } + + /** @deprecated this method should not be called */ + async subscribe(_options: EventsServiceSubscribeOptions): Promise { + throw new NotImplementedError(); } } diff --git a/plugins/events-node/src/service.ts b/plugins/events-node/src/service.ts index af0e562c80..8d754f6957 100644 --- a/plugins/events-node/src/service.ts +++ b/plugins/events-node/src/service.ts @@ -38,11 +38,18 @@ export const eventsServiceFactory = createServiceFactory({ deps: { pluginMetadata: coreServices.pluginMetadata, rootLogger: coreServices.rootLogger, + discovery: coreServices.discovery, + logger: coreServices.logger, + auth: coreServices.auth, }, async createRootContext({ rootLogger }) { return DefaultEventsService.create({ logger: rootLogger }); }, - async factory({ pluginMetadata }, eventsService) { - return eventsService.forPlugin(pluginMetadata.getId()); + async factory({ pluginMetadata, discovery, logger, auth }, eventsService) { + return eventsService.forPlugin(pluginMetadata.getId(), { + discovery, + logger, + auth, + }); }, });