From 361ca74ac20262fab9e98e87c9ddefdad8ed65ea Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Fri, 9 Dec 2022 16:34:15 -0800 Subject: [PATCH 01/14] Added a mechanism to perform out-of-sequence deltas Signed-off-by: Damon Kaswell --- .../package.json | 1 + .../src/engine/IncrementalIngestionEngine.ts | 47 +- .../src/module/WrapperProviders.ts | 10 +- ...talIngestionEntityProviderCatalogModule.ts | 2 +- .../src/router/routes.ts | 441 ++++++++++-------- .../src/service/IncrementalCatalogBuilder.ts | 10 +- .../src/types.ts | 14 +- 7 files changed, 314 insertions(+), 211 deletions(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/package.json b/plugins/catalog-backend-module-incremental-ingestion/package.json index 15826cd65a..25252e74f4 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/package.json +++ b/plugins/catalog-backend-module-incremental-ingestion/package.json @@ -40,6 +40,7 @@ "@backstage/catalog-model": "workspace:^", "@backstage/config": "workspace:^", "@backstage/errors": "workspace:^", + "@backstage/plugin-events-node": "workspace:^", "@backstage/plugin-catalog-backend": "workspace:^", "@backstage/plugin-catalog-node": "workspace:^", "@backstage/plugin-permission-common": "workspace:^", diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts index e2000c609e..ece2c5f574 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts @@ -21,14 +21,18 @@ import { performance } from 'perf_hooks'; import { Duration, DurationObjectUnits } from 'luxon'; import { v4 } from 'uuid'; import { stringifyError } from '@backstage/errors'; +import { EventParams, EventSubscriber } from '@backstage/plugin-events-node'; -export class IncrementalIngestionEngine implements IterationEngine { +export class IncrementalIngestionEngine + implements IterationEngine, EventSubscriber +{ private readonly restLength: Duration; private readonly backoff: DurationObjectUnits[]; + private readonly providerEventTopic: string; private manager: IncrementalIngestionDatabaseManager; - constructor(private options: IterationEngineOptions) { + constructor(private options: IterationEngineOptions) { this.manager = options.manager; this.restLength = Duration.fromObject(options.restLength); this.backoff = options.backoff ?? [ @@ -37,6 +41,7 @@ export class IncrementalIngestionEngine implements IterationEngine { { minutes: 30 }, { hours: 3 }, ]; + this.providerEventTopic = `${options.provider.getProviderName()}-delta`; } async taskFn(signal: AbortSignal) { @@ -326,4 +331,42 @@ export class IncrementalIngestionEngine implements IterationEngine { removed, }); } + + async onEvent(params: EventParams): Promise { + const { topic, eventPayload } = params; + if (topic !== this.providerEventTopic) { + return; + } + + const { logger, provider, connection } = this.options; + logger.info( + `incremental-engine: Received ${this.providerEventTopic} event`, + ); + + const payload = eventPayload as TInput; + + if (!provider.deltaMapper) { + return; + } + + const update = provider.deltaMapper(payload); + + if (update.delta) { + await connection.applyMutation({ + type: 'delta', + ...update.delta, + }); + logger.info( + `incremental-engine: Processed ${this.providerEventTopic} event`, + ); + } else { + logger.info( + `incremental-engine: Rejected ${this.providerEventTopic} event - empty or invalid`, + ); + } + } + + supportsEventTopics(): string[] { + return [this.providerEventTopic]; + } } 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 44da5067fd..ef8ac07b04 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts @@ -28,7 +28,7 @@ import { Duration } from 'luxon'; import { IncrementalIngestionDatabaseManager } from '../database/IncrementalIngestionDatabaseManager'; import { applyDatabaseMigrations } from '../database/migrations'; import { IncrementalIngestionEngine } from '../engine/IncrementalIngestionEngine'; -import { createIncrementalProviderRouter } from '../router/routes'; +import { IncrementalProviderRouter } from '../router/routes'; import { IncrementalEntityProvider, IncrementalEntityProviderOptions, @@ -72,14 +72,14 @@ export class WrapperProviders { } async adminRouter(): Promise { - return createIncrementalProviderRouter( + return await new IncrementalProviderRouter( new IncrementalIngestionDatabaseManager({ client: this.options.client }), loggerToWinstonLogger(this.options.logger), - ); + ).createRouter(); } - private async startProvider( - provider: IncrementalEntityProvider, + private async startProvider( + provider: IncrementalEntityProvider, providerOptions: IncrementalEntityProviderOptions, connection: EntityProviderConnection, ) { 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 0ecb45292c..2ed7dda728 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.ts @@ -38,7 +38,7 @@ export const incrementalIngestionEntityProviderCatalogModule = env, options: { providers: Array<{ - provider: IncrementalEntityProvider; + provider: IncrementalEntityProvider; options: IncrementalEntityProviderOptions; }>; }, diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/router/routes.ts b/plugins/catalog-backend-module-incremental-ingestion/src/router/routes.ts index 9f480e879e..d03619d5a4 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/router/routes.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/router/routes.ts @@ -15,218 +15,269 @@ */ import { errorHandler } from '@backstage/backend-common'; +import { stringifyError } from '@backstage/errors'; +import { EventBroker, EventPublisher } from '@backstage/plugin-events-node'; import express from 'express'; import Router from 'express-promise-router'; import { Logger } from 'winston'; import { IncrementalIngestionDatabaseManager } from '../database/IncrementalIngestionDatabaseManager'; import { PROVIDER_BASE_PATH, PROVIDER_CLEANUP, PROVIDER_HEALTH } from './paths'; -export const createIncrementalProviderRouter = async ( - manager: IncrementalIngestionDatabaseManager, - logger: Logger, -) => { - const router = Router(); - router.use(express.json()); +export class IncrementalProviderRouter implements EventPublisher { + private manager: IncrementalIngestionDatabaseManager; + private logger: Logger; + private eventBroker: EventBroker | undefined; - // Get the overall health of all incremental providers - router.get(PROVIDER_HEALTH, async (_, res) => { - const records = await manager.healthcheck(); - const providers = records.map(record => record.provider_name); - const duplicates = [ - ...new Set(providers.filter((e, i, a) => a.indexOf(e) !== i)), - ]; + constructor(manager: IncrementalIngestionDatabaseManager, logger: Logger) { + this.manager = manager; + this.logger = logger; + } - if (duplicates.length > 0) { - res.json({ healthy: false, duplicateIngestions: duplicates }); - } else { - res.json({ healthy: true }); - } - }); + async setEventBroker(eventBroker: EventBroker): Promise { + this.eventBroker = eventBroker; + } - // Clean up and pause all providers - router.post(PROVIDER_CLEANUP, async (_, res) => { - const result = await manager.cleanupProviders(); - res.json(result); - }); + async createRouter() { + const router = Router(); + router.use(express.json()); - // Get basic status of the provider - router.get(PROVIDER_BASE_PATH, async (req, res) => { - const { provider } = req.params; - const record = await manager.getCurrentIngestionRecord(provider); - if (record) { - res.json({ - success: true, - status: { - current_action: record.status, - next_action_at: new Date(record.next_action_at), - }, - last_error: record.last_error, - }); - } else { - const providers: string[] = await manager.listProviders(); - if (providers.includes(provider)) { + // Get the overall health of all incremental providers + router.get(PROVIDER_HEALTH, async (_, res) => { + const records = await this.manager.healthcheck(); + const providers = records.map(record => record.provider_name); + const duplicates = [ + ...new Set(providers.filter((e, i, a) => a.indexOf(e) !== i)), + ]; + + if (duplicates.length > 0) { + res.json({ healthy: false, duplicateIngestions: duplicates }); + } else { + res.json({ healthy: true }); + } + }); + + // Clean up and pause all providers + router.post(PROVIDER_CLEANUP, async (_, res) => { + const result = await this.manager.cleanupProviders(); + res.json(result); + }); + + // Get basic status of the provider + router.get(PROVIDER_BASE_PATH, async (req, res) => { + const { provider } = req.params; + const record = await this.manager.getCurrentIngestionRecord(provider); + if (record) { res.json({ success: true, status: { - current_action: 'rest complete, waiting to start', + current_action: record.status, + next_action_at: new Date(record.next_action_at), }, + last_error: record.last_error, }); } else { - logger.error( - `${provider} - No ingestion record found in the database!`, - ); - res.status(404).json({ - success: false, - status: {}, - last_error: `Provider '${provider}' not found`, - }); + const providers: string[] = await this.manager.listProviders(); + if (providers.includes(provider)) { + res.json({ + success: true, + status: { + current_action: 'rest complete, waiting to start', + }, + }); + } else { + this.logger.error( + `${provider} - No ingestion record found in the database!`, + ); + res.status(404).json({ + success: false, + status: {}, + last_error: `Provider '${provider}' not found`, + }); + } } - } - }); - - // Trigger the provider's next action - router.post(`${PROVIDER_BASE_PATH}/trigger`, async (req, res) => { - const { provider } = req.params; - const record = await manager.getCurrentIngestionRecord(provider); - if (record) { - await manager.triggerNextProviderAction(provider); - res.json({ - success: true, - message: `${provider}: Next action triggered.`, - }); - } else { - const providers: string[] = await manager.listProviders(); - if (providers.includes(provider)) { - logger.debug(`${provider} - Ingestion record found`); - res.json({ - success: true, - message: 'Unable to trigger next action (provider is restarting)', - }); - } else { - res.status(404).json({ - success: false, - message: `Provider '${provider}' not found`, - }); - } - } - }); - - // Start a brand-new ingestion cycle for the provider. - // (Cancel's the current run if active, or marks it complete if resting) - router.post(`${PROVIDER_BASE_PATH}/start`, async (req, res) => { - const { provider } = req.params; - - const record = await manager.getCurrentIngestionRecord(provider); - if (record) { - const ingestionId = record.id; - if (record.status === 'resting') { - await manager.setProviderComplete(ingestionId); - } else { - await manager.setProviderCanceling(ingestionId); - } - res.json({ - success: true, - message: `${provider}: Next cycle triggered.`, - }); - } else { - const providers: string[] = await manager.listProviders(); - if (providers.includes(provider)) { - logger.debug(`${provider} - Ingestion record found`); - res.json({ - success: true, - message: 'Provider is already restarting', - }); - } else { - res.status(404).json({ - success: false, - message: `Provider '${provider}' not found`, - }); - } - } - }); - - // Stop the provider and pause it for 24 hours - router.post(`${PROVIDER_BASE_PATH}/cancel`, async (req, res) => { - const { provider } = req.params; - const record = await manager.getCurrentIngestionRecord(provider); - if (record) { - const next_action_at = new Date(); - next_action_at.setTime(next_action_at.getTime() + 24 * 60 * 60 * 1000); - await manager.updateByName(provider, { - next_action: 'nothing (done)', - ingestion_completed_at: new Date(), - next_action_at, - status: 'resting', - }); - res.json({ - success: true, - message: `${provider}: Current ingestion canceled.`, - }); - } else { - const providers: string[] = await manager.listProviders(); - if (providers.includes(provider)) { - logger.debug(`${provider} - Ingestion record found`); - res.json({ - success: true, - message: 'Provider is currently restarting, please wait.', - }); - } else { - res.status(404).json({ - success: false, - message: `Provider '${provider}' not found`, - }); - } - } - }); - - // Wipe out all ingestion records for the provider and pause for 24 hours - router.delete(PROVIDER_BASE_PATH, async (req, res) => { - const { provider } = req.params; - const result = await manager.purgeAndResetProvider(provider); - res.json(result); - }); - - // Get the ingestion marks for the current cycle - router.get(`${PROVIDER_BASE_PATH}/marks`, async (req, res) => { - const { provider } = req.params; - const record = await manager.getCurrentIngestionRecord(provider); - if (record) { - const id = record.id; - const records = await manager.getAllMarks(id); - res.json({ success: true, records }); - } else { - const providers: string[] = await manager.listProviders(); - if (providers.includes(provider)) { - logger.debug(`${provider} - Ingestion record found`); - res.json({ - success: true, - message: 'No records yet (provider is restarting)', - }); - } else { - logger.error( - `${provider} - No ingestion record found in the database!`, - ); - res.status(404).json({ - success: false, - status: {}, - last_error: `Provider '${provider}' not found`, - }); - } - } - }); - - router.delete(`${PROVIDER_BASE_PATH}/marks`, async (req, res) => { - const { provider } = req.params; - const deletions = await manager.clearFinishedIngestions(provider); - - res.json({ - success: true, - message: `Expired marks for provider '${provider}' removed.`, - deletions, }); - }); - router.use(errorHandler()); + // Trigger the provider's next action + router.post(`${PROVIDER_BASE_PATH}/trigger`, async (req, res) => { + const { provider } = req.params; + const record = await this.manager.getCurrentIngestionRecord(provider); + if (record) { + await this.manager.triggerNextProviderAction(provider); + res.json({ + success: true, + message: `${provider}: Next action triggered.`, + }); + } else { + const providers: string[] = await this.manager.listProviders(); + if (providers.includes(provider)) { + this.logger.debug(`${provider} - Ingestion record found`); + res.json({ + success: true, + message: 'Unable to trigger next action (provider is restarting)', + }); + } else { + res.status(404).json({ + success: false, + message: `Provider '${provider}' not found`, + }); + } + } + }); - return router; -}; + // Start a brand-new ingestion cycle for the provider. + // (Cancel's the current run if active, or marks it complete if resting) + router.post(`${PROVIDER_BASE_PATH}/start`, async (req, res) => { + const { provider } = req.params; + + const record = await this.manager.getCurrentIngestionRecord(provider); + if (record) { + const ingestionId = record.id; + if (record.status === 'resting') { + await this.manager.setProviderComplete(ingestionId); + } else { + await this.manager.setProviderCanceling(ingestionId); + } + res.json({ + success: true, + message: `${provider}: Next cycle triggered.`, + }); + } else { + const providers: string[] = await this.manager.listProviders(); + if (providers.includes(provider)) { + this.logger.debug(`${provider} - Ingestion record found`); + res.json({ + success: true, + message: 'Provider is already restarting', + }); + } else { + res.status(404).json({ + success: false, + message: `Provider '${provider}' not found`, + }); + } + } + }); + + // Stop the provider and pause it for 24 hours + router.post(`${PROVIDER_BASE_PATH}/cancel`, async (req, res) => { + const { provider } = req.params; + const record = await this.manager.getCurrentIngestionRecord(provider); + if (record) { + const next_action_at = new Date(); + next_action_at.setTime(next_action_at.getTime() + 24 * 60 * 60 * 1000); + await this.manager.updateByName(provider, { + next_action: 'nothing (done)', + ingestion_completed_at: new Date(), + next_action_at, + status: 'resting', + }); + res.json({ + success: true, + message: `${provider}: Current ingestion canceled.`, + }); + } else { + const providers: string[] = await this.manager.listProviders(); + if (providers.includes(provider)) { + this.logger.debug(`${provider} - Ingestion record found`); + res.json({ + success: true, + message: 'Provider is currently restarting, please wait.', + }); + } else { + res.status(404).json({ + success: false, + message: `Provider '${provider}' not found`, + }); + } + } + }); + + // Wipe out all ingestion records for the provider and pause for 24 hours + router.delete(PROVIDER_BASE_PATH, async (req, res) => { + const { provider } = req.params; + const result = await this.manager.purgeAndResetProvider(provider); + res.json(result); + }); + + // Get the ingestion marks for the current cycle + router.get(`${PROVIDER_BASE_PATH}/marks`, async (req, res) => { + const { provider } = req.params; + const record = await this.manager.getCurrentIngestionRecord(provider); + if (record) { + const id = record.id; + const records = await this.manager.getAllMarks(id); + res.json({ success: true, records }); + } else { + const providers: string[] = await this.manager.listProviders(); + if (providers.includes(provider)) { + this.logger.debug(`${provider} - Ingestion record found`); + res.json({ + success: true, + message: 'No records yet (provider is restarting)', + }); + } else { + this.logger.error( + `${provider} - No ingestion record found in the database!`, + ); + res.status(404).json({ + success: false, + status: {}, + last_error: `Provider '${provider}' not found`, + }); + } + } + }); + + router.delete(`${PROVIDER_BASE_PATH}/marks`, async (req, res) => { + const { provider } = req.params; + const deletions = await this.manager.clearFinishedIngestions(provider); + + res.json({ + success: true, + message: `Expired marks for provider '${provider}' removed.`, + deletions, + }); + }); + + router.post(`${PROVIDER_BASE_PATH}/delta`, async (req, res) => { + const { provider } = req.params; + + const topic = `${provider}-push`; + + const eventPayload = req.body; + + if (!this.eventBroker) { + res.status(500).json({ + success: false, + provider, + message: `The payload could not be processed!`, + }); + throw new Error('Event broker not initialized!'); + } + + try { + await this.eventBroker.publish({ + topic, + eventPayload, + }); + res.json({ + success: true, + provider, + message: 'Payload submitted.', + }); + } catch (e) { + res.status(500).json({ + success: false, + provider, + message: `There was an error submitting the payload: ${stringifyError( + e, + )}`, + }); + } + }); + + router.use(errorHandler()); + + return router; + } +} diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/service/IncrementalCatalogBuilder.ts b/plugins/catalog-backend-module-incremental-ingestion/src/service/IncrementalCatalogBuilder.ts index d6bc5549aa..e65698d4a4 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/service/IncrementalCatalogBuilder.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/service/IncrementalCatalogBuilder.ts @@ -24,7 +24,7 @@ import { Knex } from 'knex'; import { IncrementalIngestionEngine } from '../engine/IncrementalIngestionEngine'; import { applyDatabaseMigrations } from '../database/migrations'; import { IncrementalIngestionDatabaseManager } from '../database/IncrementalIngestionDatabaseManager'; -import { createIncrementalProviderRouter } from '../router/routes'; +import { IncrementalProviderRouter } from '../router/routes'; import { Deferred } from '../util'; /** @public */ @@ -60,16 +60,16 @@ export class IncrementalCatalogBuilder { router: 'IncrementalProviderAdmin', }); - const incrementalAdminRouter = await createIncrementalProviderRouter( + const incrementalAdminRouter = await new IncrementalProviderRouter( this.manager, routerLogger, - ); + ).createRouter(); return { incrementalAdminRouter }; } - addIncrementalEntityProvider( - provider: IncrementalEntityProvider, + addIncrementalEntityProvider( + provider: IncrementalEntityProvider, options: IncrementalEntityProviderOptions, ) { const { burstInterval, burstLength, restLength } = options; diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts index 51a8738500..79563fd2e8 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts @@ -46,7 +46,7 @@ import { IncrementalIngestionDatabaseManager } from './database/IncrementalInges * * @public */ -export interface IncrementalEntityProvider { +export interface IncrementalEntityProvider { /** * This name must be unique between all of the entity providers * operating in the catalog. @@ -75,6 +75,14 @@ export interface IncrementalEntityProvider { * @param burst - a function which performs a series of iterations */ around(burst: (context: TContext) => Promise): Promise; + + /** + * If present, this method maps incoming payloads to apply updates + * outside of the incremental ingestion schedule. + */ + deltaMapper?: (payload: TInput) => { + delta: { added: DeferredEntity[]; removed: DeferredEntity[] } | undefined; + }; } /** @@ -154,11 +162,11 @@ export interface IterationEngine { taskFn: TaskFunction; } -export interface IterationEngineOptions { +export interface IterationEngineOptions { logger: Logger; connection: EntityProviderConnection; manager: IncrementalIngestionDatabaseManager; - provider: IncrementalEntityProvider; + provider: IncrementalEntityProvider; restLength: DurationObjectUnits; ready: Promise; backoff?: IncrementalEntityProviderOptions['backoff']; From c10f27b44cece704e1488808a685b043df1043ad Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Fri, 9 Dec 2022 16:56:15 -0800 Subject: [PATCH 02/14] Fix test and module Signed-off-by: Damon Kaswell --- .../api-report.md | 16 ++++++++++++---- ...lIngestionEntityProviderCatalogModule.test.ts | 2 +- .../src/run.ts | 2 +- 3 files changed, 14 insertions(+), 6 deletions(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/api-report.md b/plugins/catalog-backend-module-incremental-ingestion/api-report.md index 387e505c33..11ffdd565c 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/api-report.md +++ b/plugins/catalog-backend-module-incremental-ingestion/api-report.md @@ -33,8 +33,8 @@ export type EntityIteratorResult = // @public (undocumented) export class IncrementalCatalogBuilder { // (undocumented) - addIncrementalEntityProvider( - provider: IncrementalEntityProvider, + addIncrementalEntityProvider( + provider: IncrementalEntityProvider, options: IncrementalEntityProviderOptions, ): void; // (undocumented) @@ -48,8 +48,16 @@ export class IncrementalCatalogBuilder { } // @public -export interface IncrementalEntityProvider { +export interface IncrementalEntityProvider { around(burst: (context: TContext) => Promise): Promise; + deltaMapper?: (payload: TInput) => { + delta: + | { + added: DeferredEntity[]; + removed: DeferredEntity[]; + } + | undefined; + }; getProviderName(): string; next( context: TContext, @@ -70,7 +78,7 @@ export interface IncrementalEntityProviderOptions { // @alpha export const incrementalIngestionEntityProviderCatalogModule: (options: { providers: { - provider: IncrementalEntityProvider; + provider: IncrementalEntityProvider; options: IncrementalEntityProviderOptions; }[]; }) => BackendFeature; 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 index 06cbcb010e..d196693db8 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.test.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.test.ts @@ -24,7 +24,7 @@ import { incrementalIngestionEntityProviderCatalogModule } from './incrementalIn describe('bitbucketServerEntityProviderCatalogModule', () => { it('should register provider at the catalog extension point', async () => { - const provider1: IncrementalEntityProvider = { + const provider1: IncrementalEntityProvider = { getProviderName: () => 'provider1', around: burst => burst(0), next: async (cursor, _context) => { diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/run.ts b/plugins/catalog-backend-module-incremental-ingestion/src/run.ts index 08d95f4417..2dee5716c0 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/run.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/run.ts @@ -36,7 +36,7 @@ import { incrementalIngestionEntityProviderCatalogModule, } from '.'; -const provider: IncrementalEntityProvider = { +const provider: IncrementalEntityProvider = { getProviderName: () => 'test-provider', around: burst => burst(0), next: async (_context, cursor) => { From 1ba120faa3483413627f4cac4767580f0d3fa4b6 Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Fri, 9 Dec 2022 16:58:48 -0800 Subject: [PATCH 03/14] Update changelog Signed-off-by: Damon Kaswell --- .changeset/bright-eagles-love.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changeset/bright-eagles-love.md diff --git a/.changeset/bright-eagles-love.md b/.changeset/bright-eagles-love.md new file mode 100644 index 0000000000..eade94f6f8 --- /dev/null +++ b/.changeset/bright-eagles-love.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend-module-incremental-ingestion': minor +--- + +Added new mechanism to handle deltas in incremental providers From 034830f54ff4a1bbdfb26f9792a0d95e04962c71 Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Sat, 10 Dec 2022 05:58:41 -0800 Subject: [PATCH 04/14] Update yarn.lock Signed-off-by: Damon Kaswell --- .../catalog-backend-module-incremental-ingestion/package.json | 2 +- yarn.lock | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/package.json b/plugins/catalog-backend-module-incremental-ingestion/package.json index 25252e74f4..ace6cfebdb 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/package.json +++ b/plugins/catalog-backend-module-incremental-ingestion/package.json @@ -40,9 +40,9 @@ "@backstage/catalog-model": "workspace:^", "@backstage/config": "workspace:^", "@backstage/errors": "workspace:^", - "@backstage/plugin-events-node": "workspace:^", "@backstage/plugin-catalog-backend": "workspace:^", "@backstage/plugin-catalog-node": "workspace:^", + "@backstage/plugin-events-node": "workspace:^", "@backstage/plugin-permission-common": "workspace:^", "@types/express": "^4.17.6", "@types/luxon": "^3.0.0", diff --git a/yarn.lock b/yarn.lock index 33e5718702..4bf126ba7f 100644 --- a/yarn.lock +++ b/yarn.lock @@ -5109,6 +5109,7 @@ __metadata: "@backstage/errors": "workspace:^" "@backstage/plugin-catalog-backend": "workspace:^" "@backstage/plugin-catalog-node": "workspace:^" + "@backstage/plugin-events-node": "workspace:^" "@backstage/plugin-permission-common": "workspace:^" "@types/express": ^4.17.6 "@types/luxon": ^3.0.0 From e88d16ee91632aa21b7f9043732b8037c85805f1 Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Thu, 15 Dec 2022 09:06:14 -0800 Subject: [PATCH 05/14] Synchronize ingestion_mark_entities table with deltas Signed-off-by: Damon Kaswell --- .../IncrementalIngestionDatabaseManager.ts | 26 +++++++++++++++++ .../src/engine/IncrementalIngestionEngine.ts | 29 +++++++++++++++++++ .../src/types.ts | 7 ++++- 3 files changed, 61 insertions(+), 1 deletion(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts index 9eb52c4558..bb319b311e 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts @@ -269,6 +269,17 @@ export class IncrementalIngestionDatabaseManager { }); } + /** + * This method is used to remove entity records from the ingestion_mark_entities + * table by their entity reference. + */ + async deleteEntityRecordsByRef(entities: { entityRef: string }[]) { + const refs = entities.map(e => e.entityRef); + await this.client.transaction(async tx => { + await tx('ingestion_mark_entities').delete().whereIn('ref', refs); + }); + } + /** * Creates a new ingestion record. * @param provider - string @@ -535,6 +546,21 @@ export class IncrementalIngestionDatabaseManager { }); } + /** + * Returns the first record from `ingestion_marks` for the supplied ingestionId. + * @param ingestionId - string + * @returns MarkRecord | undefined + */ + async getFirstMark(ingestionId: string) { + return await this.client.transaction(async tx => { + const mark = await tx('ingestion_marks') + .where('ingestion_id', ingestionId) + .orderBy('sequence', 'asc') + .first(); + return mark; + }); + } + async getAllMarks(ingestionId: string) { return await this.client.transaction(async tx => { const marks = await tx('ingestion_marks') diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts index ece2c5f574..f0507dbbc6 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts @@ -339,6 +339,7 @@ export class IncrementalIngestionEngine } const { logger, provider, connection } = this.options; + const providerName = provider.getProviderName(); logger.info( `incremental-engine: Received ${this.providerEventTopic} event`, ); @@ -352,6 +353,34 @@ export class IncrementalIngestionEngine const update = provider.deltaMapper(payload); if (update.delta) { + if (update.delta.added.length > 0) { + const ingestionRecord = await this.manager.getCurrentIngestionRecord( + providerName, + ); + + if (!ingestionRecord) { + logger.debug( + `incremental-engine: Skipping delta addition because incremental ingestion is restarting.`, + ); + } else { + const mark = + ingestionRecord.status === 'resting' + ? await this.manager.getLastMark(ingestionRecord.id) + : await this.manager.getFirstMark(ingestionRecord.id); + + if (!mark) { + throw new Error( + `Cannot apply delta, page records are missing! Please re-run incremental ingestion for ${providerName}.`, + ); + } + await this.manager.createMarkEntities(mark.id, update.delta.added); + } + } + + if (update.delta.removed.length > 0) { + await this.manager.deleteEntityRecordsByRef(update.delta.removed); + } + await connection.applyMutation({ type: 'delta', ...update.delta, diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts index 79563fd2e8..f1325f5cc9 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts @@ -81,7 +81,12 @@ export interface IncrementalEntityProvider { * outside of the incremental ingestion schedule. */ deltaMapper?: (payload: TInput) => { - delta: { added: DeferredEntity[]; removed: DeferredEntity[] } | undefined; + delta: + | { + added: DeferredEntity[]; + removed: { entityRef: string }[]; + } + | undefined; }; } From 7fd3b629df1753de8c7e4de187fa74223da0811a Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Thu, 15 Dec 2022 09:12:47 -0800 Subject: [PATCH 06/14] Formatting and api report Signed-off-by: Damon Kaswell --- .../api-report.md | 4 +++- .../src/engine/IncrementalIngestionEngine.ts | 2 +- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/api-report.md b/plugins/catalog-backend-module-incremental-ingestion/api-report.md index 11ffdd565c..c56ae0e7bf 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/api-report.md +++ b/plugins/catalog-backend-module-incremental-ingestion/api-report.md @@ -54,7 +54,9 @@ export interface IncrementalEntityProvider { delta: | { added: DeferredEntity[]; - removed: DeferredEntity[]; + removed: { + entityRef: string; + }[]; } | undefined; }; diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts index f0507dbbc6..5f5df5ffe1 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts @@ -389,7 +389,7 @@ export class IncrementalIngestionEngine `incremental-engine: Processed ${this.providerEventTopic} event`, ); } else { - logger.info( + logger.warn( `incremental-engine: Rejected ${this.providerEventTopic} event - empty or invalid`, ); } From bcb0ff4f90a3bdb9a70789ec97bbdec4ee72b311 Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Thu, 12 Jan 2023 08:02:29 -0800 Subject: [PATCH 07/14] Simplify type params where they aren't needed Signed-off-by: Damon Kaswell --- .../src/engine/IncrementalIngestionEngine.ts | 8 +++---- .../src/module/WrapperProviders.test.ts | 4 ++-- .../src/module/WrapperProviders.ts | 6 +++--- ...gestionEntityProviderCatalogModule.test.ts | 2 +- ...talIngestionEntityProviderCatalogModule.ts | 2 +- .../src/run.ts | 4 ++-- .../src/service/IncrementalCatalogBuilder.ts | 4 ++-- .../src/types.ts | 21 ++++++++----------- 8 files changed, 23 insertions(+), 28 deletions(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts index 6213dcf786..e4d11b5f0d 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts @@ -23,7 +23,7 @@ import { v4 } from 'uuid'; import { stringifyError } from '@backstage/errors'; import { EventParams, EventSubscriber } from '@backstage/plugin-events-node'; -export class IncrementalIngestionEngine +export class IncrementalIngestionEngine implements IterationEngine, EventSubscriber { private readonly restLength: Duration; @@ -32,7 +32,7 @@ export class IncrementalIngestionEngine private manager: IncrementalIngestionDatabaseManager; - constructor(private options: IterationEngineOptions) { + constructor(private options: IterationEngineOptions) { this.manager = options.manager; this.restLength = Duration.fromObject(options.restLength); this.backoff = options.backoff ?? [ @@ -347,13 +347,11 @@ export class IncrementalIngestionEngine `incremental-engine: Received ${this.providerEventTopic} event`, ); - const payload = eventPayload as TInput; - if (!provider.deltaMapper) { return; } - const update = provider.deltaMapper(payload); + const update = provider.deltaMapper(eventPayload); if (update.delta) { if (update.delta.added.length > 0) { 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 0d910f664d..500d30d992 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 @@ -41,7 +41,7 @@ describe('WrapperProviders', () => { async databaseId => { const client = await databases.init(databaseId); - const provider1: IncrementalEntityProvider = { + const provider1: IncrementalEntityProvider = { getProviderName: () => 'provider1', around: burst => burst(0), next: async (_context, cursor) => { @@ -51,7 +51,7 @@ describe('WrapperProviders', () => { }, }; - const provider2: IncrementalEntityProvider = { + const provider2: IncrementalEntityProvider = { getProviderName: () => 'provider2', around: burst => burst(0), next: async (_context, cursor) => { 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 2308db681b..6c82928bbf 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts @@ -58,7 +58,7 @@ export class WrapperProviders { ) {} wrap( - provider: IncrementalEntityProvider, + provider: IncrementalEntityProvider, options: IncrementalEntityProviderOptions, ): EntityProvider { this.numberOfProvidersToConnect += 1; @@ -81,8 +81,8 @@ export class WrapperProviders { ).createRouter(); } - private async startProvider( - provider: IncrementalEntityProvider, + private async startProvider( + provider: IncrementalEntityProvider, providerOptions: IncrementalEntityProviderOptions, connection: EntityProviderConnection, ) { 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 index d196693db8..06dfd1467c 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.test.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.test.ts @@ -24,7 +24,7 @@ import { incrementalIngestionEntityProviderCatalogModule } from './incrementalIn describe('bitbucketServerEntityProviderCatalogModule', () => { it('should register provider at the catalog extension point', async () => { - const provider1: IncrementalEntityProvider = { + const provider1: IncrementalEntityProvider = { getProviderName: () => 'provider1', around: burst => burst(0), next: async (cursor, _context) => { 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 2ed7dda728..68f8bf1c90 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.ts @@ -38,7 +38,7 @@ export const incrementalIngestionEntityProviderCatalogModule = env, options: { providers: Array<{ - provider: IncrementalEntityProvider; + provider: IncrementalEntityProvider; options: IncrementalEntityProviderOptions; }>; }, diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/run.ts b/plugins/catalog-backend-module-incremental-ingestion/src/run.ts index 2dee5716c0..0f28dd7004 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/run.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/run.ts @@ -36,10 +36,10 @@ import { incrementalIngestionEntityProviderCatalogModule, } from '.'; -const provider: IncrementalEntityProvider = { +const provider: IncrementalEntityProvider = { getProviderName: () => 'test-provider', around: burst => burst(0), - next: async (_context, cursor) => { + next: async (_context, cursor: number | undefined) => { await new Promise(resolve => setTimeout(resolve, 500)); if (cursor === undefined || cursor < 3) { console.log(`### Returning batch #${cursor}`); diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/service/IncrementalCatalogBuilder.ts b/plugins/catalog-backend-module-incremental-ingestion/src/service/IncrementalCatalogBuilder.ts index e65698d4a4..875e77a7e9 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/service/IncrementalCatalogBuilder.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/service/IncrementalCatalogBuilder.ts @@ -68,8 +68,8 @@ export class IncrementalCatalogBuilder { return { incrementalAdminRouter }; } - addIncrementalEntityProvider( - provider: IncrementalEntityProvider, + addIncrementalEntityProvider( + provider: IncrementalEntityProvider, options: IncrementalEntityProviderOptions, ) { const { burstInterval, burstLength, restLength } = options; diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts index f1325f5cc9..acd5371573 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts @@ -46,7 +46,7 @@ import { IncrementalIngestionDatabaseManager } from './database/IncrementalInges * * @public */ -export interface IncrementalEntityProvider { +export interface IncrementalEntityProvider { /** * This name must be unique between all of the entity providers * operating in the catalog. @@ -62,10 +62,7 @@ export interface IncrementalEntityProvider { * @returns The entities to be ingested, as well as the cursor of * the next page after this one. */ - next( - context: TContext, - cursor?: TCursor, - ): Promise>; + next(context: unknown, cursor?: unknown): Promise; /** * Do any setup and teardown necessary in order to provide the @@ -74,13 +71,13 @@ export interface IncrementalEntityProvider { * * @param burst - a function which performs a series of iterations */ - around(burst: (context: TContext) => Promise): Promise; + around(burst: (context: unknown) => Promise): Promise; /** * If present, this method maps incoming payloads to apply updates * outside of the incremental ingestion schedule. */ - deltaMapper?: (payload: TInput) => { + deltaMapper?: (payload: unknown) => { delta: | { added: DeferredEntity[]; @@ -96,16 +93,16 @@ export interface IncrementalEntityProvider { * * @public */ -export type EntityIteratorResult = +export type EntityIteratorResult = | { done: false; entities: DeferredEntity[]; - cursor: T; + cursor: unknown; } | { done: true; entities?: DeferredEntity[]; - cursor?: T; + cursor?: unknown; }; /** @public */ @@ -167,11 +164,11 @@ export interface IterationEngine { taskFn: TaskFunction; } -export interface IterationEngineOptions { +export interface IterationEngineOptions { logger: Logger; connection: EntityProviderConnection; manager: IncrementalIngestionDatabaseManager; - provider: IncrementalEntityProvider; + provider: IncrementalEntityProvider; restLength: DurationObjectUnits; ready: Promise; backoff?: IncrementalEntityProviderOptions['backoff']; From d5415fedea08d3746ed40329e6a3d8b8ceecc839 Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Thu, 12 Jan 2023 08:06:32 -0800 Subject: [PATCH 08/14] Update API report Signed-off-by: Damon Kaswell --- .../api-report.md | 23 ++++++++----------- 1 file changed, 10 insertions(+), 13 deletions(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/api-report.md b/plugins/catalog-backend-module-incremental-ingestion/api-report.md index c56ae0e7bf..aa526a8fe8 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/api-report.md +++ b/plugins/catalog-backend-module-incremental-ingestion/api-report.md @@ -18,23 +18,23 @@ import { Router } from 'express'; import type { UrlReader } from '@backstage/backend-common'; // @public -export type EntityIteratorResult = +export type EntityIteratorResult = | { done: false; entities: DeferredEntity[]; - cursor: T; + cursor: unknown; } | { done: true; entities?: DeferredEntity[]; - cursor?: T; + cursor?: unknown; }; // @public (undocumented) export class IncrementalCatalogBuilder { // (undocumented) - addIncrementalEntityProvider( - provider: IncrementalEntityProvider, + addIncrementalEntityProvider( + provider: IncrementalEntityProvider, options: IncrementalEntityProviderOptions, ): void; // (undocumented) @@ -48,9 +48,9 @@ export class IncrementalCatalogBuilder { } // @public -export interface IncrementalEntityProvider { - around(burst: (context: TContext) => Promise): Promise; - deltaMapper?: (payload: TInput) => { +export interface IncrementalEntityProvider { + around(burst: (context: unknown) => Promise): Promise; + deltaMapper?: (payload: unknown) => { delta: | { added: DeferredEntity[]; @@ -61,10 +61,7 @@ export interface IncrementalEntityProvider { | undefined; }; getProviderName(): string; - next( - context: TContext, - cursor?: TCursor, - ): Promise>; + next(context: unknown, cursor?: unknown): Promise; } // @public (undocumented) @@ -80,7 +77,7 @@ export interface IncrementalEntityProviderOptions { // @alpha export const incrementalIngestionEntityProviderCatalogModule: (options: { providers: { - provider: IncrementalEntityProvider; + provider: IncrementalEntityProvider; options: IncrementalEntityProviderOptions; }[]; }) => BackendFeature; From 66d70c7d3c8450c3e2efd675c2fdba87ec1a4b3b Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Thu, 12 Jan 2023 09:38:14 -0800 Subject: [PATCH 09/14] Revert "Simplify type params where they aren't needed" This reverts commit bcb0ff4f90a3bdb9a70789ec97bbdec4ee72b311. Signed-off-by: Damon Kaswell --- .../src/engine/IncrementalIngestionEngine.ts | 8 ++++--- .../src/module/WrapperProviders.test.ts | 4 ++-- .../src/module/WrapperProviders.ts | 6 +++--- ...gestionEntityProviderCatalogModule.test.ts | 2 +- ...talIngestionEntityProviderCatalogModule.ts | 2 +- .../src/run.ts | 4 ++-- .../src/service/IncrementalCatalogBuilder.ts | 4 ++-- .../src/types.ts | 21 +++++++++++-------- 8 files changed, 28 insertions(+), 23 deletions(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts index e4d11b5f0d..6213dcf786 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts @@ -23,7 +23,7 @@ import { v4 } from 'uuid'; import { stringifyError } from '@backstage/errors'; import { EventParams, EventSubscriber } from '@backstage/plugin-events-node'; -export class IncrementalIngestionEngine +export class IncrementalIngestionEngine implements IterationEngine, EventSubscriber { private readonly restLength: Duration; @@ -32,7 +32,7 @@ export class IncrementalIngestionEngine private manager: IncrementalIngestionDatabaseManager; - constructor(private options: IterationEngineOptions) { + constructor(private options: IterationEngineOptions) { this.manager = options.manager; this.restLength = Duration.fromObject(options.restLength); this.backoff = options.backoff ?? [ @@ -347,11 +347,13 @@ export class IncrementalIngestionEngine `incremental-engine: Received ${this.providerEventTopic} event`, ); + const payload = eventPayload as TInput; + if (!provider.deltaMapper) { return; } - const update = provider.deltaMapper(eventPayload); + const update = provider.deltaMapper(payload); if (update.delta) { if (update.delta.added.length > 0) { 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 500d30d992..0d910f664d 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 @@ -41,7 +41,7 @@ describe('WrapperProviders', () => { async databaseId => { const client = await databases.init(databaseId); - const provider1: IncrementalEntityProvider = { + const provider1: IncrementalEntityProvider = { getProviderName: () => 'provider1', around: burst => burst(0), next: async (_context, cursor) => { @@ -51,7 +51,7 @@ describe('WrapperProviders', () => { }, }; - const provider2: IncrementalEntityProvider = { + const provider2: IncrementalEntityProvider = { getProviderName: () => 'provider2', around: burst => burst(0), next: async (_context, cursor) => { 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 6c82928bbf..2308db681b 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts @@ -58,7 +58,7 @@ export class WrapperProviders { ) {} wrap( - provider: IncrementalEntityProvider, + provider: IncrementalEntityProvider, options: IncrementalEntityProviderOptions, ): EntityProvider { this.numberOfProvidersToConnect += 1; @@ -81,8 +81,8 @@ export class WrapperProviders { ).createRouter(); } - private async startProvider( - provider: IncrementalEntityProvider, + private async startProvider( + provider: IncrementalEntityProvider, providerOptions: IncrementalEntityProviderOptions, connection: EntityProviderConnection, ) { 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 index 06dfd1467c..d196693db8 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.test.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.test.ts @@ -24,7 +24,7 @@ import { incrementalIngestionEntityProviderCatalogModule } from './incrementalIn describe('bitbucketServerEntityProviderCatalogModule', () => { it('should register provider at the catalog extension point', async () => { - const provider1: IncrementalEntityProvider = { + const provider1: IncrementalEntityProvider = { getProviderName: () => 'provider1', around: burst => burst(0), next: async (cursor, _context) => { 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 68f8bf1c90..2ed7dda728 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.ts @@ -38,7 +38,7 @@ export const incrementalIngestionEntityProviderCatalogModule = env, options: { providers: Array<{ - provider: IncrementalEntityProvider; + provider: IncrementalEntityProvider; options: IncrementalEntityProviderOptions; }>; }, diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/run.ts b/plugins/catalog-backend-module-incremental-ingestion/src/run.ts index 0f28dd7004..2dee5716c0 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/run.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/run.ts @@ -36,10 +36,10 @@ import { incrementalIngestionEntityProviderCatalogModule, } from '.'; -const provider: IncrementalEntityProvider = { +const provider: IncrementalEntityProvider = { getProviderName: () => 'test-provider', around: burst => burst(0), - next: async (_context, cursor: number | undefined) => { + next: async (_context, cursor) => { await new Promise(resolve => setTimeout(resolve, 500)); if (cursor === undefined || cursor < 3) { console.log(`### Returning batch #${cursor}`); diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/service/IncrementalCatalogBuilder.ts b/plugins/catalog-backend-module-incremental-ingestion/src/service/IncrementalCatalogBuilder.ts index 875e77a7e9..e65698d4a4 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/service/IncrementalCatalogBuilder.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/service/IncrementalCatalogBuilder.ts @@ -68,8 +68,8 @@ export class IncrementalCatalogBuilder { return { incrementalAdminRouter }; } - addIncrementalEntityProvider( - provider: IncrementalEntityProvider, + addIncrementalEntityProvider( + provider: IncrementalEntityProvider, options: IncrementalEntityProviderOptions, ) { const { burstInterval, burstLength, restLength } = options; diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts index acd5371573..f1325f5cc9 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts @@ -46,7 +46,7 @@ import { IncrementalIngestionDatabaseManager } from './database/IncrementalInges * * @public */ -export interface IncrementalEntityProvider { +export interface IncrementalEntityProvider { /** * This name must be unique between all of the entity providers * operating in the catalog. @@ -62,7 +62,10 @@ export interface IncrementalEntityProvider { * @returns The entities to be ingested, as well as the cursor of * the next page after this one. */ - next(context: unknown, cursor?: unknown): Promise; + next( + context: TContext, + cursor?: TCursor, + ): Promise>; /** * Do any setup and teardown necessary in order to provide the @@ -71,13 +74,13 @@ export interface IncrementalEntityProvider { * * @param burst - a function which performs a series of iterations */ - around(burst: (context: unknown) => Promise): Promise; + around(burst: (context: TContext) => Promise): Promise; /** * If present, this method maps incoming payloads to apply updates * outside of the incremental ingestion schedule. */ - deltaMapper?: (payload: unknown) => { + deltaMapper?: (payload: TInput) => { delta: | { added: DeferredEntity[]; @@ -93,16 +96,16 @@ export interface IncrementalEntityProvider { * * @public */ -export type EntityIteratorResult = +export type EntityIteratorResult = | { done: false; entities: DeferredEntity[]; - cursor: unknown; + cursor: T; } | { done: true; entities?: DeferredEntity[]; - cursor?: unknown; + cursor?: T; }; /** @public */ @@ -164,11 +167,11 @@ export interface IterationEngine { taskFn: TaskFunction; } -export interface IterationEngineOptions { +export interface IterationEngineOptions { logger: Logger; connection: EntityProviderConnection; manager: IncrementalIngestionDatabaseManager; - provider: IncrementalEntityProvider; + provider: IncrementalEntityProvider; restLength: DurationObjectUnits; ready: Promise; backoff?: IncrementalEntityProviderOptions['backoff']; From a319a5977408f8cee44f81548c2313df4e7569ce Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Thu, 12 Jan 2023 09:54:40 -0800 Subject: [PATCH 10/14] One more try on types Signed-off-by: Damon Kaswell --- .../api-report.md | 21 +++++++++++-------- .../src/engine/IncrementalIngestionEngine.ts | 8 +++---- .../src/module/WrapperProviders.ts | 4 ++-- ...gestionEntityProviderCatalogModule.test.ts | 2 +- ...talIngestionEntityProviderCatalogModule.ts | 2 +- .../src/run.ts | 2 +- .../src/service/IncrementalCatalogBuilder.ts | 4 ++-- .../src/types.ts | 8 +++---- 8 files changed, 26 insertions(+), 25 deletions(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/api-report.md b/plugins/catalog-backend-module-incremental-ingestion/api-report.md index aa526a8fe8..c79d07ad26 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/api-report.md +++ b/plugins/catalog-backend-module-incremental-ingestion/api-report.md @@ -18,23 +18,23 @@ import { Router } from 'express'; import type { UrlReader } from '@backstage/backend-common'; // @public -export type EntityIteratorResult = +export type EntityIteratorResult = | { done: false; entities: DeferredEntity[]; - cursor: unknown; + cursor: T; } | { done: true; entities?: DeferredEntity[]; - cursor?: unknown; + cursor?: T; }; // @public (undocumented) export class IncrementalCatalogBuilder { // (undocumented) - addIncrementalEntityProvider( - provider: IncrementalEntityProvider, + addIncrementalEntityProvider( + provider: IncrementalEntityProvider, options: IncrementalEntityProviderOptions, ): void; // (undocumented) @@ -48,8 +48,8 @@ export class IncrementalCatalogBuilder { } // @public -export interface IncrementalEntityProvider { - around(burst: (context: unknown) => Promise): Promise; +export interface IncrementalEntityProvider { + around(burst: (context: TContext) => Promise): Promise; deltaMapper?: (payload: unknown) => { delta: | { @@ -61,7 +61,10 @@ export interface IncrementalEntityProvider { | undefined; }; getProviderName(): string; - next(context: unknown, cursor?: unknown): Promise; + next( + context: TContext, + cursor?: TCursor, + ): Promise>; } // @public (undocumented) @@ -77,7 +80,7 @@ export interface IncrementalEntityProviderOptions { // @alpha export const incrementalIngestionEntityProviderCatalogModule: (options: { providers: { - provider: IncrementalEntityProvider; + provider: IncrementalEntityProvider; options: IncrementalEntityProviderOptions; }[]; }) => BackendFeature; diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts index 6213dcf786..e4d11b5f0d 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts @@ -23,7 +23,7 @@ import { v4 } from 'uuid'; import { stringifyError } from '@backstage/errors'; import { EventParams, EventSubscriber } from '@backstage/plugin-events-node'; -export class IncrementalIngestionEngine +export class IncrementalIngestionEngine implements IterationEngine, EventSubscriber { private readonly restLength: Duration; @@ -32,7 +32,7 @@ export class IncrementalIngestionEngine private manager: IncrementalIngestionDatabaseManager; - constructor(private options: IterationEngineOptions) { + constructor(private options: IterationEngineOptions) { this.manager = options.manager; this.restLength = Duration.fromObject(options.restLength); this.backoff = options.backoff ?? [ @@ -347,13 +347,11 @@ export class IncrementalIngestionEngine `incremental-engine: Received ${this.providerEventTopic} event`, ); - const payload = eventPayload as TInput; - if (!provider.deltaMapper) { return; } - const update = provider.deltaMapper(payload); + const update = provider.deltaMapper(eventPayload); if (update.delta) { if (update.delta.added.length > 0) { 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 2308db681b..e714568c4b 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts @@ -81,8 +81,8 @@ export class WrapperProviders { ).createRouter(); } - private async startProvider( - provider: IncrementalEntityProvider, + private async startProvider( + provider: IncrementalEntityProvider, providerOptions: IncrementalEntityProviderOptions, connection: EntityProviderConnection, ) { 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 index d196693db8..06cbcb010e 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.test.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.test.ts @@ -24,7 +24,7 @@ import { incrementalIngestionEntityProviderCatalogModule } from './incrementalIn describe('bitbucketServerEntityProviderCatalogModule', () => { it('should register provider at the catalog extension point', async () => { - const provider1: IncrementalEntityProvider = { + const provider1: IncrementalEntityProvider = { getProviderName: () => 'provider1', around: burst => burst(0), next: async (cursor, _context) => { 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 2ed7dda728..0ecb45292c 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/incrementalIngestionEntityProviderCatalogModule.ts @@ -38,7 +38,7 @@ export const incrementalIngestionEntityProviderCatalogModule = env, options: { providers: Array<{ - provider: IncrementalEntityProvider; + provider: IncrementalEntityProvider; options: IncrementalEntityProviderOptions; }>; }, diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/run.ts b/plugins/catalog-backend-module-incremental-ingestion/src/run.ts index 2dee5716c0..08d95f4417 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/run.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/run.ts @@ -36,7 +36,7 @@ import { incrementalIngestionEntityProviderCatalogModule, } from '.'; -const provider: IncrementalEntityProvider = { +const provider: IncrementalEntityProvider = { getProviderName: () => 'test-provider', around: burst => burst(0), next: async (_context, cursor) => { diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/service/IncrementalCatalogBuilder.ts b/plugins/catalog-backend-module-incremental-ingestion/src/service/IncrementalCatalogBuilder.ts index e65698d4a4..6d2b3e2be5 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/service/IncrementalCatalogBuilder.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/service/IncrementalCatalogBuilder.ts @@ -68,8 +68,8 @@ export class IncrementalCatalogBuilder { return { incrementalAdminRouter }; } - addIncrementalEntityProvider( - provider: IncrementalEntityProvider, + addIncrementalEntityProvider( + provider: IncrementalEntityProvider, options: IncrementalEntityProviderOptions, ) { const { burstInterval, burstLength, restLength } = options; diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts index f1325f5cc9..24919e2647 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts @@ -46,7 +46,7 @@ import { IncrementalIngestionDatabaseManager } from './database/IncrementalInges * * @public */ -export interface IncrementalEntityProvider { +export interface IncrementalEntityProvider { /** * This name must be unique between all of the entity providers * operating in the catalog. @@ -80,7 +80,7 @@ export interface IncrementalEntityProvider { * If present, this method maps incoming payloads to apply updates * outside of the incremental ingestion schedule. */ - deltaMapper?: (payload: TInput) => { + deltaMapper?: (payload: unknown) => { delta: | { added: DeferredEntity[]; @@ -167,11 +167,11 @@ export interface IterationEngine { taskFn: TaskFunction; } -export interface IterationEngineOptions { +export interface IterationEngineOptions { logger: Logger; connection: EntityProviderConnection; manager: IncrementalIngestionDatabaseManager; - provider: IncrementalEntityProvider; + provider: IncrementalEntityProvider; restLength: DurationObjectUnits; ready: Promise; backoff?: IncrementalEntityProviderOptions['backoff']; From b7154302eac7f95c7e2ea346b80a8ab494e3d080 Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Thu, 12 Jan 2023 10:37:16 -0800 Subject: [PATCH 11/14] Fix typo Signed-off-by: Damon Kaswell --- .../src/engine/IncrementalIngestionEngine.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts index e4d11b5f0d..311af0f6de 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts @@ -41,7 +41,7 @@ export class IncrementalIngestionEngine { minutes: 30 }, { hours: 3 }, ]; - this.providerEventTopic = `${options.provider.getProviderName()}-delta`; + this.providerEventTopic = `${options.provider.getProviderName()}-push`; } async taskFn(signal: AbortSignal) { From c01b1f0ac527f4f76da29acca8b98846d09f00be Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Fri, 13 Jan 2023 08:33:46 -0800 Subject: [PATCH 12/14] Rename 'deltaMapper' to 'onEvent' Signed-off-by: Damon Kaswell --- .../api-report.md | 12 ++++++------ .../src/engine/IncrementalIngestionEngine.ts | 4 ++-- .../src/types.ts | 9 ++++++--- 3 files changed, 14 insertions(+), 11 deletions(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/api-report.md b/plugins/catalog-backend-module-incremental-ingestion/api-report.md index c79d07ad26..1052afda74 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/api-report.md +++ b/plugins/catalog-backend-module-incremental-ingestion/api-report.md @@ -50,7 +50,12 @@ export class IncrementalCatalogBuilder { // @public export interface IncrementalEntityProvider { around(burst: (context: TContext) => Promise): Promise; - deltaMapper?: (payload: unknown) => { + getProviderName(): string; + next( + context: TContext, + cursor?: TCursor, + ): Promise>; + onEvent?: (payload: unknown) => { delta: | { added: DeferredEntity[]; @@ -60,11 +65,6 @@ export interface IncrementalEntityProvider { } | undefined; }; - getProviderName(): string; - next( - context: TContext, - cursor?: TCursor, - ): Promise>; } // @public (undocumented) diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts index 311af0f6de..a21092a838 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts @@ -347,11 +347,11 @@ export class IncrementalIngestionEngine `incremental-engine: Received ${this.providerEventTopic} event`, ); - if (!provider.deltaMapper) { + if (!provider.onEvent) { return; } - const update = provider.deltaMapper(eventPayload); + const update = provider.onEvent(eventPayload); if (update.delta) { if (update.delta.added.length > 0) { diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts index 24919e2647..315757d88a 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts @@ -77,10 +77,13 @@ export interface IncrementalEntityProvider { around(burst: (context: TContext) => Promise): Promise; /** - * If present, this method maps incoming payloads to apply updates - * outside of the incremental ingestion schedule. + * If present, this method accepts an incoming event, and maps the + * payload to an object containing a delta mutation. + * + * If a delta is present, the incremental entity provider will apply + * it automatically. */ - deltaMapper?: (payload: unknown) => { + onEvent?: (payload: unknown) => { delta: | { added: DeferredEntity[]; From 70febd490d052d9f1ec45491308e39665f9c9e0f Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Tue, 17 Jan 2023 07:46:15 -0800 Subject: [PATCH 13/14] Better event handling Signed-off-by: Damon Kaswell --- .../api-report.md | 23 +++--- .../src/engine/IncrementalIngestionEngine.ts | 74 ++++++++++--------- .../src/router/routes.ts | 2 +- .../src/types.ts | 28 ++++--- 4 files changed, 68 insertions(+), 59 deletions(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/api-report.md b/plugins/catalog-backend-module-incremental-ingestion/api-report.md index 1052afda74..3fc932ad2d 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/api-report.md +++ b/plugins/catalog-backend-module-incremental-ingestion/api-report.md @@ -10,6 +10,7 @@ import { CatalogBuilder } from '@backstage/plugin-catalog-backend'; import type { Config } from '@backstage/config'; import type { DeferredEntity } from '@backstage/plugin-catalog-backend'; import type { DurationObjectUnits } from 'luxon'; +import { EventParams } from '@backstage/plugin-events-node'; import type { Logger } from 'winston'; import type { PermissionEvaluator } from '@backstage/plugin-permission-common'; import type { PluginDatabaseManager } from '@backstage/backend-common'; @@ -55,16 +56,18 @@ export interface IncrementalEntityProvider { context: TContext, cursor?: TCursor, ): Promise>; - onEvent?: (payload: unknown) => { - delta: - | { - added: DeferredEntity[]; - removed: { - entityRef: string; - }[]; - } - | undefined; - }; + onEvent?: (params: EventParams) => + | { + delta: + | { + added: DeferredEntity[]; + removed: { + entityRef: string; + }[]; + } + | undefined; + } + | undefined; } // @public (undocumented) diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts index a21092a838..c1a62b7a73 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts @@ -336,14 +336,14 @@ export class IncrementalIngestionEngine } async onEvent(params: EventParams): Promise { - const { topic, eventPayload } = params; + const { topic } = params; if (topic !== this.providerEventTopic) { return; } const { logger, provider, connection } = this.options; const providerName = provider.getProviderName(); - logger.info( + logger.debug( `incremental-engine: Received ${this.providerEventTopic} event`, ); @@ -351,48 +351,50 @@ export class IncrementalIngestionEngine return; } - const update = provider.onEvent(eventPayload); + const update = provider.onEvent(params); - if (update.delta) { - if (update.delta.added.length > 0) { - const ingestionRecord = await this.manager.getCurrentIngestionRecord( - providerName, - ); - - if (!ingestionRecord) { - logger.debug( - `incremental-engine: Skipping delta addition because incremental ingestion is restarting.`, + if (update) { + if (update.delta) { + if (update.delta.added.length > 0) { + const ingestionRecord = await this.manager.getCurrentIngestionRecord( + providerName, ); - } else { - const mark = - ingestionRecord.status === 'resting' - ? await this.manager.getLastMark(ingestionRecord.id) - : await this.manager.getFirstMark(ingestionRecord.id); - if (!mark) { - throw new Error( - `Cannot apply delta, page records are missing! Please re-run incremental ingestion for ${providerName}.`, + if (!ingestionRecord) { + logger.debug( + `incremental-engine: Skipping delta addition because incremental ingestion is restarting.`, ); + } else { + const mark = + ingestionRecord.status === 'resting' + ? await this.manager.getLastMark(ingestionRecord.id) + : await this.manager.getFirstMark(ingestionRecord.id); + + if (!mark) { + throw new Error( + `Cannot apply delta, page records are missing! Please re-run incremental ingestion for ${providerName}.`, + ); + } + await this.manager.createMarkEntities(mark.id, update.delta.added); } - await this.manager.createMarkEntities(mark.id, update.delta.added); } - } - if (update.delta.removed.length > 0) { - await this.manager.deleteEntityRecordsByRef(update.delta.removed); - } + if (update.delta.removed.length > 0) { + await this.manager.deleteEntityRecordsByRef(update.delta.removed); + } - await connection.applyMutation({ - type: 'delta', - ...update.delta, - }); - logger.info( - `incremental-engine: Processed ${this.providerEventTopic} event`, - ); - } else { - logger.warn( - `incremental-engine: Rejected ${this.providerEventTopic} event - empty or invalid`, - ); + await connection.applyMutation({ + type: 'delta', + ...update.delta, + }); + logger.debug( + `incremental-engine: Processed ${this.providerEventTopic} event`, + ); + } else { + logger.warn( + `incremental-engine: Rejected ${this.providerEventTopic} event - empty or invalid`, + ); + } } } diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/router/routes.ts b/plugins/catalog-backend-module-incremental-ingestion/src/router/routes.ts index d03619d5a4..b3e3ae5792 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/router/routes.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/router/routes.ts @@ -239,7 +239,7 @@ export class IncrementalProviderRouter implements EventPublisher { }); }); - router.post(`${PROVIDER_BASE_PATH}/delta`, async (req, res) => { + router.post(`${PROVIDER_BASE_PATH}/event`, async (req, res) => { const { provider } = req.params; const topic = `${provider}-push`; diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts index 315757d88a..efadee3a76 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts @@ -27,6 +27,7 @@ import type { DeferredEntity, EntityProviderConnection, } from '@backstage/plugin-catalog-backend'; +import { EventParams } from '@backstage/plugin-events-node'; import type { PermissionEvaluator } from '@backstage/plugin-permission-common'; import type { DurationObjectUnits } from 'luxon'; import type { Logger } from 'winston'; @@ -77,20 +78,23 @@ export interface IncrementalEntityProvider { around(burst: (context: TContext) => Promise): Promise; /** - * If present, this method accepts an incoming event, and maps the - * payload to an object containing a delta mutation. + * This method accepts an incoming event for the provider, and + * (optionally) maps the payload to an object containing a delta + * mutation. * - * If a delta is present, the incremental entity provider will apply - * it automatically. + * If a valid delta is returned by this method, it will be ingested + * automatically by the provider. */ - onEvent?: (payload: unknown) => { - delta: - | { - added: DeferredEntity[]; - removed: { entityRef: string }[]; - } - | undefined; - }; + onEvent?: (params: EventParams) => + | { + delta: + | { + added: DeferredEntity[]; + removed: { entityRef: string }[]; + } + | undefined; + } + | undefined; } /** From 3c1eae932492b2adb5a11e0556a958d912a4f7e9 Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Thu, 19 Jan 2023 08:44:30 -0800 Subject: [PATCH 14/14] Do not create new event route, rely on events backend instead Signed-off-by: Damon Kaswell --- .../api-report.md | 23 +++-- .../src/engine/IncrementalIngestionEngine.ts | 88 +++++++++---------- .../src/module/WrapperProviders.ts | 4 +- .../src/router/routes.ts | 46 +--------- .../src/types.ts | 42 +++++---- 5 files changed, 84 insertions(+), 119 deletions(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/api-report.md b/plugins/catalog-backend-module-incremental-ingestion/api-report.md index 3fc932ad2d..4bf6bb9456 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/api-report.md +++ b/plugins/catalog-backend-module-incremental-ingestion/api-report.md @@ -51,23 +51,22 @@ export class IncrementalCatalogBuilder { // @public export interface IncrementalEntityProvider { around(burst: (context: TContext) => Promise): Promise; + eventHandler?: { + onEvent: (params: EventParams) => + | undefined + | { + added: DeferredEntity[]; + removed: { + entityRef: string; + }[]; + }; + supportsEventTopics: () => string[]; + }; getProviderName(): string; next( context: TContext, cursor?: TCursor, ): Promise>; - onEvent?: (params: EventParams) => - | { - delta: - | { - added: DeferredEntity[]; - removed: { - entityRef: string; - }[]; - } - | undefined; - } - | undefined; } // @public (undocumented) diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts index c1a62b7a73..228c7a2505 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts @@ -28,7 +28,6 @@ export class IncrementalIngestionEngine { private readonly restLength: Duration; private readonly backoff: DurationObjectUnits[]; - private readonly providerEventTopic: string; private manager: IncrementalIngestionDatabaseManager; @@ -41,7 +40,6 @@ export class IncrementalIngestionEngine { minutes: 30 }, { hours: 3 }, ]; - this.providerEventTopic = `${options.provider.getProviderName()}-push`; } async taskFn(signal: AbortSignal) { @@ -337,68 +335,68 @@ export class IncrementalIngestionEngine async onEvent(params: EventParams): Promise { const { topic } = params; - if (topic !== this.providerEventTopic) { + if (!this.supportsEventTopics().includes(topic)) { return; } const { logger, provider, connection } = this.options; const providerName = provider.getProviderName(); - logger.debug( - `incremental-engine: Received ${this.providerEventTopic} event`, - ); + logger.debug(`incremental-engine: ${providerName} received ${topic} event`); - if (!provider.onEvent) { + if (!provider.eventHandler) { return; } - const update = provider.onEvent(params); + const delta = provider.eventHandler.onEvent(params); - if (update) { - if (update.delta) { - if (update.delta.added.length > 0) { - const ingestionRecord = await this.manager.getCurrentIngestionRecord( - providerName, + if (delta) { + if (delta.added.length > 0) { + const ingestionRecord = await this.manager.getCurrentIngestionRecord( + providerName, + ); + + if (!ingestionRecord) { + logger.debug( + `incremental-engine: ${providerName} skipping delta addition because incremental ingestion is restarting.`, ); + } else { + const mark = + ingestionRecord.status === 'resting' + ? await this.manager.getLastMark(ingestionRecord.id) + : await this.manager.getFirstMark(ingestionRecord.id); - if (!ingestionRecord) { - logger.debug( - `incremental-engine: Skipping delta addition because incremental ingestion is restarting.`, + if (!mark) { + throw new Error( + `Cannot apply delta, page records are missing! Please re-run incremental ingestion for ${providerName}.`, ); - } else { - const mark = - ingestionRecord.status === 'resting' - ? await this.manager.getLastMark(ingestionRecord.id) - : await this.manager.getFirstMark(ingestionRecord.id); - - if (!mark) { - throw new Error( - `Cannot apply delta, page records are missing! Please re-run incremental ingestion for ${providerName}.`, - ); - } - await this.manager.createMarkEntities(mark.id, update.delta.added); } + await this.manager.createMarkEntities(mark.id, delta.added); } - - if (update.delta.removed.length > 0) { - await this.manager.deleteEntityRecordsByRef(update.delta.removed); - } - - await connection.applyMutation({ - type: 'delta', - ...update.delta, - }); - logger.debug( - `incremental-engine: Processed ${this.providerEventTopic} event`, - ); - } else { - logger.warn( - `incremental-engine: Rejected ${this.providerEventTopic} event - empty or invalid`, - ); } + + if (delta.removed.length > 0) { + await this.manager.deleteEntityRecordsByRef(delta.removed); + } + + await connection.applyMutation({ + type: 'delta', + ...delta, + }); + logger.debug( + `incremental-engine: ${providerName} processed delta from '${topic}' event`, + ); + } else { + logger.warn( + `incremental-engine: Rejected delta from '${topic}' event - empty or invalid`, + ); } } supportsEventTopics(): string[] { - return [this.providerEventTopic]; + const { provider } = this.options; + const topics = provider.eventHandler + ? provider.eventHandler.supportsEventTopics() + : []; + return topics; } } 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 e714568c4b..4738893c95 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts @@ -81,8 +81,8 @@ export class WrapperProviders { ).createRouter(); } - private async startProvider( - provider: IncrementalEntityProvider, + private async startProvider( + provider: IncrementalEntityProvider, providerOptions: IncrementalEntityProviderOptions, connection: EntityProviderConnection, ) { diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/router/routes.ts b/plugins/catalog-backend-module-incremental-ingestion/src/router/routes.ts index b3e3ae5792..fc26abdf32 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/router/routes.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/router/routes.ts @@ -15,28 +15,21 @@ */ import { errorHandler } from '@backstage/backend-common'; -import { stringifyError } from '@backstage/errors'; -import { EventBroker, EventPublisher } from '@backstage/plugin-events-node'; import express from 'express'; import Router from 'express-promise-router'; import { Logger } from 'winston'; import { IncrementalIngestionDatabaseManager } from '../database/IncrementalIngestionDatabaseManager'; import { PROVIDER_BASE_PATH, PROVIDER_CLEANUP, PROVIDER_HEALTH } from './paths'; -export class IncrementalProviderRouter implements EventPublisher { +export class IncrementalProviderRouter { private manager: IncrementalIngestionDatabaseManager; private logger: Logger; - private eventBroker: EventBroker | undefined; constructor(manager: IncrementalIngestionDatabaseManager, logger: Logger) { this.manager = manager; this.logger = logger; } - async setEventBroker(eventBroker: EventBroker): Promise { - this.eventBroker = eventBroker; - } - async createRouter() { const router = Router(); router.use(express.json()); @@ -239,43 +232,6 @@ export class IncrementalProviderRouter implements EventPublisher { }); }); - router.post(`${PROVIDER_BASE_PATH}/event`, async (req, res) => { - const { provider } = req.params; - - const topic = `${provider}-push`; - - const eventPayload = req.body; - - if (!this.eventBroker) { - res.status(500).json({ - success: false, - provider, - message: `The payload could not be processed!`, - }); - throw new Error('Event broker not initialized!'); - } - - try { - await this.eventBroker.publish({ - topic, - eventPayload, - }); - res.json({ - success: true, - provider, - message: 'Payload submitted.', - }); - } catch (e) { - res.status(500).json({ - success: false, - provider, - message: `There was an error submitting the payload: ${stringifyError( - e, - )}`, - }); - } - }); - router.use(errorHandler()); return router; diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts index efadee3a76..cd1e63b044 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts @@ -78,23 +78,35 @@ export interface IncrementalEntityProvider { around(burst: (context: TContext) => Promise): Promise; /** - * This method accepts an incoming event for the provider, and - * (optionally) maps the payload to an object containing a delta - * mutation. + * If set, the IncrementalEntityProvider will receive and respond to + * events. * - * If a valid delta is returned by this method, it will be ingested - * automatically by the provider. + * This system acts as a wrapper for the Backstage events bus, and + * requires the events backend to function. It does not provide its + * own events backend. See {@link https://github.com/backstage/backstage/tree/master/plugins/events-backend}. */ - onEvent?: (params: EventParams) => - | { - delta: - | { - added: DeferredEntity[]; - removed: { entityRef: string }[]; - } - | undefined; - } - | undefined; + eventHandler?: { + /** + * This method accepts an incoming event for the provider, and + * optionally maps the payload to an object containing a delta + * mutation. + * + * If a valid delta is returned by this method, it will be ingested + * automatically by the provider. + */ + onEvent: (params: EventParams) => + | undefined + | { + added: DeferredEntity[]; + removed: { entityRef: string }[]; + }; + + /** + * This method returns an array of topics for the IncrementalEntityProvider + * to respond to. + */ + supportsEventTopics: () => string[]; + }; } /**