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 diff --git a/plugins/catalog-backend-module-incremental-ingestion/api-report.md b/plugins/catalog-backend-module-incremental-ingestion/api-report.md index 387e505c33..4bf6bb9456 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'; @@ -50,6 +51,17 @@ 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, diff --git a/plugins/catalog-backend-module-incremental-ingestion/package.json b/plugins/catalog-backend-module-incremental-ingestion/package.json index bb2f3d1678..c0eb027ac2 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/package.json +++ b/plugins/catalog-backend-module-incremental-ingestion/package.json @@ -42,6 +42,7 @@ "@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", 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 e837f2b259..cf47942754 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 e94bc9c53f..228c7a2505 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts @@ -21,8 +21,11 @@ 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[]; @@ -329,4 +332,71 @@ export class IncrementalIngestionEngine implements IterationEngine { removed, }); } + + async onEvent(params: EventParams): Promise { + const { topic } = params; + if (!this.supportsEventTopics().includes(topic)) { + return; + } + + const { logger, provider, connection } = this.options; + const providerName = provider.getProviderName(); + logger.debug(`incremental-engine: ${providerName} received ${topic} event`); + + if (!provider.eventHandler) { + return; + } + + const delta = provider.eventHandler.onEvent(params); + + 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 (!mark) { + throw new Error( + `Cannot apply delta, page records are missing! Please re-run incremental ingestion for ${providerName}.`, + ); + } + await this.manager.createMarkEntities(mark.id, delta.added); + } + } + + 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[] { + 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 abe4cdb951..4738893c95 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts @@ -31,7 +31,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, @@ -75,10 +75,10 @@ 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( 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..fc26abdf32 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/router/routes.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/router/routes.ts @@ -21,212 +21,219 @@ 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 { + private manager: IncrementalIngestionDatabaseManager; + private logger: Logger; - // 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 createRouter() { + const router = Router(); + router.use(express.json()); - // Clean up and pause all providers - router.post(PROVIDER_CLEANUP, async (_, res) => { - const result = await manager.cleanupProviders(); - res.json(result); - }); + // 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)), + ]; - // 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)) { + 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.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..6d2b3e2be5 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,10 +60,10 @@ export class IncrementalCatalogBuilder { router: 'IncrementalProviderAdmin', }); - const incrementalAdminRouter = await createIncrementalProviderRouter( + const incrementalAdminRouter = await new IncrementalProviderRouter( this.manager, routerLogger, - ); + ).createRouter(); return { incrementalAdminRouter }; } diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts index 51a8738500..cd1e63b044 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'; @@ -75,6 +76,37 @@ export interface IncrementalEntityProvider { * @param burst - a function which performs a series of iterations */ around(burst: (context: TContext) => Promise): Promise; + + /** + * If set, the IncrementalEntityProvider will receive and respond to + * events. + * + * 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}. + */ + 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[]; + }; } /** diff --git a/yarn.lock b/yarn.lock index a3917b9e36..9fdb0bd4e0 100644 --- a/yarn.lock +++ b/yarn.lock @@ -4989,6 +4989,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