From 5f2debd2204825feb219120a6c6fced142839627 Mon Sep 17 00:00:00 2001 From: blam Date: Thu, 21 Nov 2024 12:41:37 +0100 Subject: [PATCH 1/5] chore: wire up subscriptions for new frontend system Signed-off-by: blam --- .../src/module/WrapperProviders.ts | 17 ++++++++++++++++- ...gModuleIncrementalIngestionEntityProvider.ts | 3 +++ 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts b/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts index c322d9adda..b510eb2ead 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts @@ -36,6 +36,7 @@ import { IncrementalEntityProvider, IncrementalEntityProviderOptions, } from '../types'; +import { EventsService } from '@backstage/plugin-events-node'; /** * Helps in the creation of the catalog entity providers that wrap the @@ -53,6 +54,7 @@ export class WrapperProviders { client: Knex; scheduler: SchedulerService; applyDatabaseMigrations?: typeof applyDatabaseMigrations; + events: EventsService; }, ) {} @@ -64,7 +66,7 @@ export class WrapperProviders { return { getProviderName: () => provider.getProviderName(), connect: async connection => { - await this.startProvider(provider, options, connection); + await this.startProvider(provider, options, connection, events); this.numberOfProvidersToConnect -= 1; if (this.numberOfProvidersToConnect === 0) { this.readySignal.resolve(); @@ -84,6 +86,7 @@ export class WrapperProviders { provider: IncrementalEntityProvider, providerOptions: IncrementalEntityProviderOptions, connection: EntityProviderConnection, + events: EventsService, ) { const logger = this.options.logger.child({ entityProvider: provider.getProviderName(), @@ -130,6 +133,18 @@ export class WrapperProviders { frequency, timeout: length, }); + + const topics = engine.supportsEventTopics(); + if (topics.length > 0) { + logger.info( + `Provider ${provider.getProviderName()} subscribing to events for topics: ${topics.join()}`, + ); + await events.subscribe({ + topics, + id: provider.getProviderName(), + onEvent: evt => engine.onEvent(evt), + }); + } } catch (error) { logger.warn( `Failed to initialize incremental ingestion provider ${provider.getProviderName()}, ${stringifyError( diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/module/catalogModuleIncrementalIngestionEntityProvider.ts b/plugins/catalog-backend-module-incremental-ingestion/src/module/catalogModuleIncrementalIngestionEntityProvider.ts index 8752bd3137..74244402a6 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/module/catalogModuleIncrementalIngestionEntityProvider.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/catalogModuleIncrementalIngestionEntityProvider.ts @@ -25,6 +25,7 @@ import { IncrementalEntityProviderOptions, } from '@backstage/plugin-catalog-backend-module-incremental-ingestion'; import { WrapperProviders } from './WrapperProviders'; +import { eventsServiceRef } from '@backstage/plugin-events-node'; /** * @public @@ -106,6 +107,7 @@ export const catalogModuleIncrementalIngestionEntityProvider = httpRouter: coreServices.httpRouter, logger: coreServices.logger, scheduler: coreServices.scheduler, + events: eventsServiceRef, }, async init({ catalog, @@ -122,6 +124,7 @@ export const catalogModuleIncrementalIngestionEntityProvider = logger, client, scheduler, + events, }); for (const entry of addedProviders) { From 74d0c1cffa54717726f88c15a27b53896dfa3961 Mon Sep 17 00:00:00 2001 From: blam Date: Thu, 21 Nov 2024 12:52:47 +0100 Subject: [PATCH 2/5] chore: wire up new events Signed-off-by: blam --- .../src/module/WrapperProviders.ts | 5 ++--- .../catalogModuleIncrementalIngestionEntityProvider.ts | 1 + 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts b/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts index b510eb2ead..0c0f1361cf 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts @@ -66,7 +66,7 @@ export class WrapperProviders { return { getProviderName: () => provider.getProviderName(), connect: async connection => { - await this.startProvider(provider, options, connection, events); + await this.startProvider(provider, options, connection); this.numberOfProvidersToConnect -= 1; if (this.numberOfProvidersToConnect === 0) { this.readySignal.resolve(); @@ -86,7 +86,6 @@ export class WrapperProviders { provider: IncrementalEntityProvider, providerOptions: IncrementalEntityProviderOptions, connection: EntityProviderConnection, - events: EventsService, ) { const logger = this.options.logger.child({ entityProvider: provider.getProviderName(), @@ -139,7 +138,7 @@ export class WrapperProviders { logger.info( `Provider ${provider.getProviderName()} subscribing to events for topics: ${topics.join()}`, ); - await events.subscribe({ + await this.options.events.subscribe({ topics, id: provider.getProviderName(), onEvent: evt => engine.onEvent(evt), diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/module/catalogModuleIncrementalIngestionEntityProvider.ts b/plugins/catalog-backend-module-incremental-ingestion/src/module/catalogModuleIncrementalIngestionEntityProvider.ts index 74244402a6..25ff33d6ba 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/module/catalogModuleIncrementalIngestionEntityProvider.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/catalogModuleIncrementalIngestionEntityProvider.ts @@ -116,6 +116,7 @@ export const catalogModuleIncrementalIngestionEntityProvider = httpRouter, logger, scheduler, + events, }) { const client = await database.getClient(); From 5aa44d2791d14ac00e8464b420d8b96cc54f167e Mon Sep 17 00:00:00 2001 From: blam Date: Thu, 21 Nov 2024 12:53:56 +0100 Subject: [PATCH 3/5] chore: chantgeset Signed-off-by: blam --- .changeset/rich-beers-eat.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changeset/rich-beers-eat.md diff --git a/.changeset/rich-beers-eat.md b/.changeset/rich-beers-eat.md new file mode 100644 index 0000000000..55115e929e --- /dev/null +++ b/.changeset/rich-beers-eat.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend-module-incremental-ingestion': patch +--- + +Wire up the events together in the new backend system From 6fbc3b359d31ab181d4196f75666c2abf9ab9afb Mon Sep 17 00:00:00 2001 From: blam Date: Thu, 21 Nov 2024 13:00:38 +0100 Subject: [PATCH 4/5] chore: fix mock services Signed-off-by: blam --- .../src/module/WrapperProviders.test.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.test.ts b/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.test.ts index 0f98384893..95b3f203c4 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.test.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.test.ts @@ -68,6 +68,7 @@ describe('WrapperProviders', () => { client, scheduler: scheduler as Partial as SchedulerService, applyDatabaseMigrations, + events: mockServices.events.mock(), }); const wrapped1 = providers.wrap(provider1, { burstInterval: { seconds: 1 }, From e327ec4859df64ae5da97f60718fc480be4676a7 Mon Sep 17 00:00:00 2001 From: blam Date: Thu, 21 Nov 2024 17:46:49 +0100 Subject: [PATCH 5/5] chore fix Signed-off-by: blam --- .../src/module/WrapperProviders.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts b/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts index 0c0f1361cf..738d5733d0 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts @@ -136,11 +136,13 @@ export class WrapperProviders { const topics = engine.supportsEventTopics(); if (topics.length > 0) { logger.info( - `Provider ${provider.getProviderName()} subscribing to events for topics: ${topics.join()}`, + `Provider ${provider.getProviderName()} subscribing to events for topics: ${topics.join( + ',', + )}`, ); await this.options.events.subscribe({ topics, - id: provider.getProviderName(), + id: `catalog-backend-module-incremental-ingestion:${provider.getProviderName()}`, onEvent: evt => engine.onEvent(evt), }); }