From 1520967c504c46da4c70328632a7c0f730456656 Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Wed, 30 Nov 2022 15:16:59 -0800 Subject: [PATCH] Implement configurable threshold and empty collection check Signed-off-by: Damon Kaswell --- .../api-report.md | 2 + .../IncrementalIngestionDatabaseManager.ts | 29 +++++++--- .../src/engine/IncrementalIngestionEngine.ts | 56 ++++++++++++------- .../src/types.ts | 18 ++++++ 4 files changed, 78 insertions(+), 27 deletions(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/api-report.md b/plugins/catalog-backend-module-incremental-ingestion/api-report.md index 5f77490091..aef16e1533 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/api-report.md +++ b/plugins/catalog-backend-module-incremental-ingestion/api-report.md @@ -62,6 +62,8 @@ export interface IncrementalEntityProviderOptions { backoff?: DurationObjectUnits[]; burstInterval: DurationObjectUnits; burstLength: DurationObjectUnits; + rejectEmptyEntityCollections?: boolean; + removalThreshold?: number; restLength: DurationObjectUnits; } 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 c9299ec7d9..5b1618a870 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts @@ -298,11 +298,24 @@ export class IncrementalIngestionDatabaseManager { * @param ingestionId - string * @returns All entities to remove for this burst. */ - async computeRemoved(provider: string) { + async computeRemoved(provider: string, providerId: string) { const previousIngestion = await this.getPreviousIngestionRecord(provider); return await this.client.transaction(async tx => { + const count = await tx('ingestion_mark_entities') + .count({ total: 'ingestion_mark_entities.ref' }) + .join( + 'ingestion_marks', + 'ingestion_marks.id', + 'ingestion_mark_entities.ingestion_mark_id', + ) + .join('ingestions', 'ingestions.id', 'ingestion_marks.ingestion_id') + .where('ingestions.id', providerId); + + const total = count.reduce((acc, cur) => acc + (cur.total as number), 0); + + const removed: { entityRef: string }[] = []; if (previousIngestion) { - const rows: { ref: string }[] = await tx('ingestion_mark_entities') + const stale: { ref: string }[] = await tx('ingestion_mark_entities') .select('ingestion_mark_entities.ref') .join( 'ingestion_marks', @@ -312,14 +325,14 @@ export class IncrementalIngestionDatabaseManager { .join('ingestions', 'ingestions.id', 'ingestion_marks.ingestion_id') .where('ingestions.id', previousIngestion.id); - const removed: { entityRef: string }[] = rows.map(e => { - return { entityRef: e.ref }; - }); - const total = rows.length ?? 0; - return { removed, total }; + removed.push( + ...stale.map(e => { + return { entityRef: e.ref }; + }), + ); } - return { removed: [], total: 0 }; + return { total, removed }; }); } 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 65deb92ad1..8df74b8891 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts @@ -22,8 +22,6 @@ import { Duration, DurationObjectUnits } from 'luxon'; import { v4 } from 'uuid'; import { stringifyError } from '@backstage/errors'; -const REMOVAL_THRESHOLD = 5; - export class IncrementalIngestionEngine implements IterationEngine { private readonly restLength: Duration; private readonly backoff: DurationObjectUnits[]; @@ -279,26 +277,46 @@ export class IncrementalIngestionEngine implements IterationEngine { ); const result = await this.manager.computeRemoved( this.options.provider.getProviderName(), + id, ); + const { total } = result; - const percentRemoved = - total > 0 ? (result.removed.length / total) * 100 : 0; - if (percentRemoved <= REMOVAL_THRESHOLD) { - this.options.logger.info( - `incremental-engine: Ingestion '${id}': Removing ${result.removed.length} entities that have no matching assets`, - ); + + let doRemoval = true; + if (this.options.rejectEmptyEntityCollections) { + if (total === 0) { + this.options.logger.error( + `incremental-engine: Ingestion '${id}': Rejecting empty entity collection!`, + ); + doRemoval = false; + } + } + + if (this.options.removalThreshold) { + // If the total entities upserted in this ingestion is 0, then + // 100% of entities are stale and marked for removal. + const percentRemoved = + total > 0 ? (result.removed.length / total) * 100 : 100; + if (percentRemoved <= this.options.removalThreshold) { + this.options.logger.info( + `incremental-engine: Ingestion '${id}': Removing ${result.removed.length} entities that have no matching assets`, + ); + } else { + const notice = `Attempted to remove ${percentRemoved}% of matching entities!`; + this.options.logger.error( + `incremental-engine: Ingestion '${id}': ${notice}`, + ); + await this.manager.updateIngestionRecordById({ + ingestionId: id, + update: { + last_error: `REMOVAL_THRESHOLD exceeded on ingestion mark ${markId}: ${notice}`, + }, + }); + doRemoval = false; + } + } + if (doRemoval) { removed.push(...result.removed); - } else { - const notice = `Attempted to remove ${percentRemoved}% of ${total} matching entities!`; - this.options.logger.error( - `incremental-engine: Ingestion '${id}': ${notice}`, - ); - await this.manager.updateIngestionRecordById({ - ingestionId: id, - update: { - last_error: `REMOVAL_THRESHOLD exceeded on ingestion mark ${markId}: ${notice}`, - }, - }); } } diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts index ec46721a51..657073b9fb 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts @@ -122,6 +122,22 @@ export interface IncrementalEntityProviderOptions { * `[{ minutes: 1}, { minutes: 5}, {minutes: 30 }, { hours: 3 }]` */ backoff?: DurationObjectUnits[]; + + /** + * If an error occurs at a data source that results in a large + * number of assets being inadvertently removed, it will result in + * Backstage removing all associated entities. To avoid that, set + * a percentage of entities past which removal will be disallowed. + */ + removalThreshold?: number; + + /** + * Similar to the removalThreshold, this option prevents removals + * in circumstances where a data source has improperly returned 0 + * assets. If set to `true`, Backstage will reject removals when + * that happens. + */ + rejectEmptyEntityCollections?: boolean; } /** @public */ @@ -146,4 +162,6 @@ export interface IterationEngineOptions { restLength: DurationObjectUnits; ready: Promise; backoff?: IncrementalEntityProviderOptions['backoff']; + removalThreshold?: number; + rejectEmptyEntityCollections?: boolean; }