From 361ca74ac20262fab9e98e87c9ddefdad8ed65ea Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Fri, 9 Dec 2022 16:34:15 -0800 Subject: [PATCH] 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'];