diff --git a/plugins/catalog-backend-module-incremental-ingestion/package.json b/plugins/catalog-backend-module-incremental-ingestion/package.json index 74ed0c8e5c..46e64465be 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/package.json +++ b/plugins/catalog-backend-module-incremental-ingestion/package.json @@ -36,6 +36,7 @@ "@backstage/backend-common": "workspace:^", "@backstage/backend-plugin-api": "workspace:^", "@backstage/backend-tasks": "workspace:^", + "@backstage/backend-test-utils": "workspace:^", "@backstage/catalog-model": "workspace:^", "@backstage/config": "workspace:^", "@backstage/errors": "workspace:^", 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 new file mode 100644 index 0000000000..5314ac769a --- /dev/null +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.test.ts @@ -0,0 +1,124 @@ +/* + * Copyright 2022 The Backstage Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { getVoidLogger } from '@backstage/backend-common'; +import { PluginTaskScheduler } from '@backstage/backend-tasks'; +import { TestDatabases } from '@backstage/backend-test-utils'; +import { ConfigReader } from '@backstage/config'; +import { Knex } from 'knex'; +import '../database/migrations'; +import { IncrementalEntityProvider } from '../types'; +import { WrapperProviders } from './WrapperProviders'; + +const applyDatabaseMigrations = jest.fn(); +jest.mock('../database/migrations', () => { + const actual = jest.requireActual('../database/migrations'); + return { + applyDatabaseMigrations: (knex: Knex) => { + applyDatabaseMigrations(knex); + return actual.applyDatabaseMigrations(knex); + }, + }; +}); + +describe('WrapperProviders', () => { + const databases = TestDatabases.create({ + ids: ['POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'], + }); + const config = new ConfigReader({}); + const logger = getVoidLogger(); + const scheduler = { + scheduleTask: jest.fn(), + }; + + beforeEach(() => { + jest.clearAllMocks(); + }); + + it.each(databases.eachSupportedId())( + 'should initialize the providers in order, %p', + async databaseId => { + const client = await databases.init(databaseId); + + const provider1: IncrementalEntityProvider<{}, number> = { + getProviderName: () => 'provider1', + around: burst => burst(0), + next: async (cursor, _context) => { + return cursor === 0 + ? { done: false, entities: [], cursor: 1 } + : { done: true }; + }, + }; + + const provider2: IncrementalEntityProvider<{}, number> = { + getProviderName: () => 'provider2', + around: burst => burst(0), + next: async (cursor, _context) => { + return cursor === 0 + ? { done: false, entities: [], cursor: 1 } + : { done: true }; + }, + }; + + const providers = new WrapperProviders({ + config, + logger, + client, + scheduler: + scheduler as Partial as PluginTaskScheduler, + }); + const wrapped1 = providers.wrap(provider1, { + burstInterval: { seconds: 1 }, + burstLength: { seconds: 1 }, + restLength: { seconds: 1 }, + }); + const wrapped2 = providers.wrap(provider2, { + burstInterval: { seconds: 1 }, + burstLength: { seconds: 1 }, + restLength: { seconds: 1 }, + }); + + let resolved = false; + (providers as any).readySignal.then(() => { + resolved = true; + }); + + expect(applyDatabaseMigrations).toHaveBeenCalledTimes(0); + expect(resolved).toBe(false); + expect(scheduler.scheduleTask).not.toHaveBeenCalled(); + + await wrapped1.connect({} as any); // simulates the catalog engine + + expect(resolved).toBe(false); + expect(applyDatabaseMigrations).toHaveBeenCalledTimes(1); + expect(scheduler.scheduleTask).toHaveBeenLastCalledWith( + expect.objectContaining({ + id: 'provider1', + }), + ); + + await wrapped2.connect({} as any); + + expect(resolved).toBe(true); + expect(applyDatabaseMigrations).toHaveBeenCalledTimes(1); + expect(scheduler.scheduleTask).toHaveBeenLastCalledWith( + expect.objectContaining({ + id: 'provider2', + }), + ); + }, + ); +}); 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 986a8fbf62..71baaae7cb 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts @@ -14,7 +14,6 @@ * limitations under the License. */ -import { PluginDatabaseManager } from '@backstage/backend-common'; import { Logger, loggerToWinstonLogger } from '@backstage/backend-plugin-api'; import { PluginTaskScheduler } from '@backstage/backend-tasks'; import { Config } from '@backstage/config'; @@ -23,6 +22,7 @@ import { EntityProvider, EntityProviderConnection, } from '@backstage/plugin-catalog-node'; +import { Knex } from 'knex'; import once from 'lodash/once'; import { Duration } from 'luxon'; import { IncrementalIngestionDatabaseManager } from '../database/IncrementalIngestionDatabaseManager'; @@ -45,10 +45,10 @@ export class WrapperProviders { private readonly readySignal = new Deferred(); constructor( - private readonly deps: { + private readonly options: { config: Config; logger: Logger; - database: PluginDatabaseManager; + client: Knex; scheduler: PluginTaskScheduler; }, ) {} @@ -58,49 +58,41 @@ export class WrapperProviders { options: IncrementalEntityProviderOptions, ): EntityProvider { this.numberOfProvidersToConnect += 1; - - const wrapper: EntityProvider = { + return { getProviderName: () => provider.getProviderName(), connect: async connection => { - this.readySignal.then(() => - this.startProvider(provider, options, connection), - ); - + await this.startProvider(provider, options, connection); this.numberOfProvidersToConnect -= 1; - if (this.numberOfProvidersToConnect === 0) { this.readySignal.resolve(); } }, }; - - return wrapper; } private async startProvider( provider: IncrementalEntityProvider, - options: IncrementalEntityProviderOptions, + providerOptions: IncrementalEntityProviderOptions, connection: EntityProviderConnection, ) { const logger = loggerToWinstonLogger( - this.deps.logger.child({ + this.options.logger.child({ entityProvider: provider.getProviderName(), }), ); try { - const client = await this.deps.database.getClient(); - await applyDatabaseMigrationsOnce(client); + await applyDatabaseMigrationsOnce(this.options.client); - const { burstInterval, burstLength, restLength } = options; + const { burstInterval, burstLength, restLength } = providerOptions; logger.info(`Connecting`); const manager = new IncrementalIngestionDatabaseManager({ - client, + client: this.options.client, }); const engine = new IncrementalIngestionEngine({ - ...options, + ...providerOptions, ready: this.readySignal, manager, logger, @@ -116,7 +108,7 @@ export class WrapperProviders { ? burstLength : Duration.fromObject(burstLength); - await this.deps.scheduler.scheduleTask({ + await this.options.scheduler.scheduleTask({ id: provider.getProviderName(), fn: engine.taskFn.bind(engine), frequency, diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.test.ts b/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.test.ts new file mode 100644 index 0000000000..d14a89e11a --- /dev/null +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.test.ts @@ -0,0 +1,80 @@ +/* + * Copyright 2022 The Backstage Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { getVoidLogger } from '@backstage/backend-common'; +import { + configServiceRef, + databaseServiceRef, + loggerServiceRef, + schedulerServiceRef, +} from '@backstage/backend-plugin-api'; +import { startTestBackend } from '@backstage/backend-test-utils'; +import { ConfigReader } from '@backstage/config'; +import { catalogProcessingExtensionPoint } from '@backstage/plugin-catalog-node'; +import { IncrementalEntityProvider } from '../types'; +import { incrementalIngestionEntityProviderCatalogModule } from './incrementalIngestionEntityProviderCatalogModule'; + +describe('bitbucketServerEntityProviderCatalogModule', () => { + it('should register provider at the catalog extension point', async () => { + const provider1: IncrementalEntityProvider<{}, number> = { + getProviderName: () => 'provider1', + around: burst => burst(0), + next: async (cursor, _context) => { + return cursor === 0 + ? { done: false, entities: [], cursor: 1 } + : { done: true }; + }, + }; + + const addEntityProvider = jest.fn(); + + const scheduler = {}; + const database = { + getClient: jest.fn(), + }; + + await startTestBackend({ + extensionPoints: [ + [catalogProcessingExtensionPoint, { addEntityProvider }], + ], + services: [ + [configServiceRef, new ConfigReader({})], + [loggerServiceRef, getVoidLogger()], + [schedulerServiceRef, scheduler], + [databaseServiceRef, database], + ], + features: [ + incrementalIngestionEntityProviderCatalogModule({ + providers: [ + { + provider: provider1, + options: { + burstInterval: { seconds: 1 }, + burstLength: { seconds: 1 }, + restLength: { seconds: 1 }, + }, + }, + ], + }), + ], + }); + + expect(addEntityProvider).toHaveBeenCalledTimes(1); + expect(addEntityProvider.mock.calls[0][0].getProviderName()).toBe( + 'provider1', + ); + }); +}); diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.ts b/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.ts index 80ac169f21..1ba75b3ed7 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.ts @@ -54,12 +54,16 @@ export const incrementalIngestionEntityProviderCatalogModule = database: databaseServiceRef, scheduler: schedulerServiceRef, }, - async init({ catalog, ...otherDeps }) { - const providers = new WrapperProviders(otherDeps); + async init({ catalog, config, logger, database, scheduler }) { + const providers = new WrapperProviders({ + config, + logger, + client: await database.getClient(), + scheduler, + }); for (const entry of options.providers) { - catalog.addEntityProvider( - providers.wrap(entry.provider, entry.options), - ); + const wrapped = providers.wrap(entry.provider, entry.options); + catalog.addEntityProvider(wrapped); } }, }); diff --git a/yarn.lock b/yarn.lock index 43cb63fa3c..6383165776 100644 --- a/yarn.lock +++ b/yarn.lock @@ -4626,6 +4626,7 @@ __metadata: "@backstage/backend-common": "workspace:^" "@backstage/backend-plugin-api": "workspace:^" "@backstage/backend-tasks": "workspace:^" + "@backstage/backend-test-utils": "workspace:^" "@backstage/catalog-model": "workspace:^" "@backstage/cli": "workspace:^" "@backstage/config": "workspace:^"