From 61d4efe9787afbe9beb0b53f0808dad45a0f6e18 Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Mon, 28 Nov 2022 07:50:31 -0800 Subject: [PATCH 01/14] Update changeset Signed-off-by: Damon Kaswell --- .changeset/sharp-shoes-bathe.md | 5 +++ .../IncrementalIngestionDatabaseManager.ts | 30 ++++++++++++----- .../src/engine/IncrementalIngestionEngine.ts | 33 ++++++++++++++++--- 3 files changed, 54 insertions(+), 14 deletions(-) create mode 100644 .changeset/sharp-shoes-bathe.md diff --git a/.changeset/sharp-shoes-bathe.md b/.changeset/sharp-shoes-bathe.md new file mode 100644 index 0000000000..7c44559e2c --- /dev/null +++ b/.changeset/sharp-shoes-bathe.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend-module-incremental-ingestion': minor +--- + +Make incremental providers more resilient to failures 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 3c91b24e2c..c6d5fe795b 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts @@ -287,6 +287,15 @@ export class IncrementalIngestionDatabaseManager { */ async computeRemoved(provider: string, ingestionId: string) { return await this.client.transaction(async tx => { + const rows = await tx('final_entities') + .count({ total: '*' }) + .join('search', 'search.entity_id', 'final_entities.entity_id') + .where( + 'search.key', + `metadata.annotations.${INCREMENTAL_ENTITY_PROVIDER_ANNOTATION}`, + ) + .andWhere('search.value', provider); + const total = rows.reduce((acc, cur) => acc + (cur.total as number), 0); const removed: { entity: string; ref: string }[] = await tx( 'final_entities', ) @@ -302,23 +311,26 @@ export class IncrementalIngestionDatabaseManager { .join('search', 'search.entity_id', 'final_entities.entity_id') .whereNotIn( 'entity_ref', - tx('ingestion_marks') + tx('ingestion.ingestion_marks') .join( - 'ingestion_mark_entities', - 'ingestion_marks.id', - 'ingestion_mark_entities.ingestion_mark_id', + 'ingestion.ingestion_mark_entities', + 'ingestion.ingestion_marks.id', + 'ingestion.ingestion_mark_entities.ingestion_mark_id', ) - .select('ingestion_mark_entities.ref') - .where('ingestion_marks.ingestion_id', ingestionId), + .select('ingestion.ingestion_mark_entities.ref') + .where('ingestion.ingestion_marks.ingestion_id', ingestionId), ) .andWhere( 'search.key', `metadata.annotations.${INCREMENTAL_ENTITY_PROVIDER_ANNOTATION}`, ) .andWhere('search.value', provider); - return removed.map(entity => { - return { entity: JSON.parse(entity.entity) }; - }); + return { + total, + removed: removed.map(entity => { + return { entity: JSON.parse(entity.entity) }; + }), + }; }); } 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 a742f63045..36b683fa9a 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts @@ -27,6 +27,8 @@ 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,12 +281,33 @@ export class IncrementalIngestionEngine implements IterationEngine { const removed: DeferredEntity[] = []; if (done) { - removed.push( - ...(await this.manager.computeRemoved( - this.options.provider.getProviderName(), - id, - )), + this.options.logger.info( + `incremental-engine: Ingestion '${id}': Final page reached, calculating removed entities`, ); + const result = await this.manager.computeRemoved( + this.options.provider.getProviderName(), + id, + ); + const total = result.total + added.length; + 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`, + ); + 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}`, + }, + }); + } } await this.options.connection.applyMutation({ From 77c207ccba33b5d80f3d06e5d5f513ddaa62f4dc Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Mon, 28 Nov 2022 08:03:35 -0800 Subject: [PATCH 02/14] Remove reference to unused schema Signed-off-by: Damon Kaswell --- .../IncrementalIngestionDatabaseManager.ts | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) 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 c6d5fe795b..0fd560544d 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts @@ -311,14 +311,14 @@ export class IncrementalIngestionDatabaseManager { .join('search', 'search.entity_id', 'final_entities.entity_id') .whereNotIn( 'entity_ref', - tx('ingestion.ingestion_marks') + tx('ingestion_marks') .join( - 'ingestion.ingestion_mark_entities', - 'ingestion.ingestion_marks.id', - 'ingestion.ingestion_mark_entities.ingestion_mark_id', + 'ingestion_mark_entities', + 'ingestion_marks.id', + 'ingestion_mark_entities.ingestion_mark_id', ) - .select('ingestion.ingestion_mark_entities.ref') - .where('ingestion.ingestion_marks.ingestion_id', ingestionId), + .select('ingestion_mark_entities.ref') + .where('ingestion_marks.ingestion_id', ingestionId), ) .andWhere( 'search.key', @@ -463,7 +463,7 @@ export class IncrementalIngestionDatabaseManager { } /** - * Starts the cancel process for the current ingestion. + * Starts the cancel process for the current * @param ingestionId - string * @param message - string (optional) */ @@ -478,7 +478,7 @@ export class IncrementalIngestionDatabaseManager { } /** - * Completes the cancel process and triggers a new ingestion. + * Completes the cancel process and triggers a new * @param ingestionId - string */ async setProviderCanceled(ingestionId: string) { From 6eefcb878076ef2324b371c4cf94cd6f369704e1 Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Mon, 28 Nov 2022 08:05:22 -0800 Subject: [PATCH 03/14] Typos Signed-off-by: Damon Kaswell --- .../src/database/IncrementalIngestionDatabaseManager.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 0fd560544d..03a749d70c 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts @@ -463,7 +463,7 @@ export class IncrementalIngestionDatabaseManager { } /** - * Starts the cancel process for the current + * Starts the cancel process for the current ingestion. * @param ingestionId - string * @param message - string (optional) */ @@ -478,7 +478,7 @@ export class IncrementalIngestionDatabaseManager { } /** - * Completes the cancel process and triggers a new + * Completes the cancel process and triggers a new ingestion. * @param ingestionId - string */ async setProviderCanceled(ingestionId: string) { From f506a945753e5b2a8045bf41a7c5ad602f4e00cb Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Mon, 28 Nov 2022 12:25:37 -0800 Subject: [PATCH 04/14] Use last ingestion cycle to generate total Signed-off-by: Damon Kaswell --- .../IncrementalIngestionDatabaseManager.ts | 45 +++++++++++++++---- 1 file changed, 36 insertions(+), 9 deletions(-) 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 03a749d70c..fd99f1207d 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts @@ -113,6 +113,25 @@ export class IncrementalIngestionDatabaseManager { }); } + /** + * Finds the last ingestion record for the named provider. + * @param provider - string + * @returns IngestionRecord | undefined + */ + async getPreviousIngestionRecord(provider: string) { + return await this.client.transaction(async tx => { + const record = await tx('ingestions') + .where('provider_name', provider) + .andWhereNot('completion_ticket', 'open') + .first(); + if (!record) { + // This is the first time this entity provider has run. Return the current record. + return await this.getCurrentIngestionRecord(provider); + } + return record; + }); + } + /** * Removes all entries from `ingestion_marks_entities`, `ingestion_marks`, and `ingestions` * for prior ingestions that completed (i.e., have a `completion_ticket` value other than 'open'). @@ -286,16 +305,24 @@ export class IncrementalIngestionDatabaseManager { * @returns All entities to remove for this burst. */ async computeRemoved(provider: string, ingestionId: string) { + const previousIngestion = (await this.getPreviousIngestionRecord( + provider, + )) as IngestionRecord; return await this.client.transaction(async tx => { - const rows = await tx('final_entities') - .count({ total: '*' }) - .join('search', 'search.entity_id', 'final_entities.entity_id') - .where( - 'search.key', - `metadata.annotations.${INCREMENTAL_ENTITY_PROVIDER_ANNOTATION}`, - ) - .andWhere('search.value', provider); - const total = rows.reduce((acc, cur) => acc + (cur.total as number), 0); + let total = 0; + if (previousIngestion.id !== ingestionId) { + const rows = await tx('ingestion_mark_entities') + .count({ total: '*' }) + .join( + 'ingestion_marks', + 'ingestion_marks.id', + 'ingestion_mark_entities.ingestion_mark_id', + ) + .join('ingestions', 'ingestions.id', 'ingestion_marks.ingestion_id') + .where('ingestions.id', previousIngestion.id); + + total = rows.reduce((acc, cur) => acc + (cur.total as number), 0); + } const removed: { entity: string; ref: string }[] = await tx( 'final_entities', ) From 3a6a5dd189fef27c59325e196aa4e3e6a34e9eed Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Tue, 29 Nov 2022 10:43:28 -0800 Subject: [PATCH 05/14] Reuse existing ingestion_mark_entity records Signed-off-by: Damon Kaswell --- .../IncrementalIngestionDatabaseManager.ts | 93 ++++++++----------- .../src/engine/IncrementalIngestionEngine.ts | 3 +- 2 files changed, 40 insertions(+), 56 deletions(-) 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 fd99f1207d..5cd8fce0be 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts @@ -16,10 +16,9 @@ import { Knex } from 'knex'; import type { DeferredEntity } from '@backstage/plugin-catalog-backend'; -import { stringifyEntityRef } from '@backstage/catalog-model'; +import { parseEntityRef, stringifyEntityRef } from '@backstage/catalog-model'; import { Duration } from 'luxon'; import { v4 } from 'uuid'; -import { INCREMENTAL_ENTITY_PROVIDER_ANNOTATION } from '../types'; import { IngestionRecord, IngestionRecordUpdate, @@ -120,15 +119,10 @@ export class IncrementalIngestionDatabaseManager { */ async getPreviousIngestionRecord(provider: string) { return await this.client.transaction(async tx => { - const record = await tx('ingestions') + return await tx('ingestions') .where('provider_name', provider) .andWhereNot('completion_ticket', 'open') .first(); - if (!record) { - // This is the first time this entity provider has run. Return the current record. - return await this.getCurrentIngestionRecord(provider); - } - return record; }); } @@ -304,15 +298,12 @@ export class IncrementalIngestionDatabaseManager { * @param ingestionId - string * @returns All entities to remove for this burst. */ - async computeRemoved(provider: string, ingestionId: string) { - const previousIngestion = (await this.getPreviousIngestionRecord( - provider, - )) as IngestionRecord; + async computeRemoved(provider: string) { + const previousIngestion = await this.getPreviousIngestionRecord(provider); return await this.client.transaction(async tx => { - let total = 0; - if (previousIngestion.id !== ingestionId) { - const rows = await tx('ingestion_mark_entities') - .count({ total: '*' }) + if (previousIngestion) { + const rows: { ref: string }[] = await tx('ingestion_mark_entities') + .select('ingestion_mark_entities.ref') .join( 'ingestion_marks', 'ingestion_marks.id', @@ -321,43 +312,23 @@ export class IncrementalIngestionDatabaseManager { .join('ingestions', 'ingestions.id', 'ingestion_marks.ingestion_id') .where('ingestions.id', previousIngestion.id); - total = rows.reduce((acc, cur) => acc + (cur.total as number), 0); + const removed: DeferredEntity[] = rows.map(e => { + const parsed = parseEntityRef(e.ref); + const entity = { + apiVersion: 'backstage.io/v1alpha1', + kind: parsed.kind, + metadata: { + name: parsed.name, + namespace: parsed.namespace, + }, + }; + return { entity }; + }); + const total = rows.length ?? 0; + return { removed, total }; } - const removed: { entity: string; ref: string }[] = await tx( - 'final_entities', - ) - .select( - tx.ref('final_entity').as('entity'), - tx.ref('refresh_state.entity_ref').as('ref'), - ) - .join( - 'refresh_state', - 'refresh_state.entity_id', - 'final_entities.entity_id', - ) - .join('search', 'search.entity_id', 'final_entities.entity_id') - .whereNotIn( - 'entity_ref', - tx('ingestion_marks') - .join( - 'ingestion_mark_entities', - 'ingestion_marks.id', - 'ingestion_mark_entities.ingestion_mark_id', - ) - .select('ingestion_mark_entities.ref') - .where('ingestion_marks.ingestion_id', ingestionId), - ) - .andWhere( - 'search.key', - `metadata.annotations.${INCREMENTAL_ENTITY_PROVIDER_ANNOTATION}`, - ) - .andWhere('search.value', provider); - return { - total, - removed: removed.map(entity => { - return { entity: JSON.parse(entity.entity) }; - }), - }; + + return { removed: [], total: 0 }; }); } @@ -585,12 +556,26 @@ export class IncrementalIngestionDatabaseManager { * @param entities - DeferredEntity[] */ async createMarkEntities(markId: string, entities: DeferredEntity[]) { + const refs = entities.map(e => stringifyEntityRef(e.entity)); + await this.client.transaction(async tx => { + const existingRefs = ( + await tx<{ ref: string }>('ingestion_mark_entities') + .select('ref') + .whereIn('ref', refs) + ).map(e => e.ref); + + const newRefs = refs.filter(e => !existingRefs.includes(e)); + + await tx('ingestion_mark_entities') + .update('ingestion_mark_id', markId) + .whereIn('ref', existingRefs); + await tx('ingestion_mark_entities').insert( - entities.map(entity => ({ + newRefs.map(ref => ({ id: v4(), ingestion_mark_id: markId, - ref: stringifyEntityRef(entity.entity), + ref, })), ); }); 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 36b683fa9a..a47cd8f58c 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts @@ -286,9 +286,8 @@ export class IncrementalIngestionEngine implements IterationEngine { ); const result = await this.manager.computeRemoved( this.options.provider.getProviderName(), - id, ); - const total = result.total + added.length; + const { total } = result; const percentRemoved = total > 0 ? (result.removed.length / total) * 100 : 0; if (percentRemoved <= REMOVAL_THRESHOLD) { From 78d5d19721e4fa219409770cef3ec1f341ce56cc Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Tue, 29 Nov 2022 14:56:00 -0800 Subject: [PATCH 06/14] Mark as patch Signed-off-by: Damon Kaswell --- .changeset/sharp-shoes-bathe.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/sharp-shoes-bathe.md b/.changeset/sharp-shoes-bathe.md index 7c44559e2c..8d3c0dfddb 100644 --- a/.changeset/sharp-shoes-bathe.md +++ b/.changeset/sharp-shoes-bathe.md @@ -1,5 +1,5 @@ --- -'@backstage/plugin-catalog-backend-module-incremental-ingestion': minor +'@backstage/plugin-catalog-backend-module-incremental-ingestion': patch --- Make incremental providers more resilient to failures From 16af3f6dacf300b9c30195e9ed0af2aa08378b33 Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Tue, 29 Nov 2022 15:18:20 -0800 Subject: [PATCH 07/14] Removed unneeded annotation Signed-off-by: Damon Kaswell --- .../README.md | 5 ++--- .../src/engine/IncrementalIngestionEngine.ts | 8 +------- .../src/types.ts | 8 -------- 3 files changed, 3 insertions(+), 18 deletions(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/README.md b/plugins/catalog-backend-module-incremental-ingestion/README.md index d14e33ea53..0a646e4857 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/README.md +++ b/plugins/catalog-backend-module-incremental-ingestion/README.md @@ -13,11 +13,10 @@ Unfortunately, these two kinds of mutations are insufficient for very large data 3. Addressing the above two use case with `full` mutation is not an option on very large datasets because a `full` mutation requires that all entities are in memory to create a diff. If your data source has 100k+ records, this can easily cause your processes to run out of memory. 4. In cases when you can use `full` mutation, committing many entities into the processing pipeline fills up the processing queue and delays the processing of entities from other entity providers. -We created the Incremental Entity Provider to address all of the above issues. The Incremental Entity Provider addresses these issues with a combination of `delta` mutations and a mark-and-sweep mechanism. Instead of doing a single `full` mutation, it performs a series of bursts. At the end of each burst, the Incremental Entity Provider performs the following three operations, +We created the Incremental Entity Provider to address all of the above issues. The Incremental Entity Provider addresses these issues with a combination of `delta` mutations and a mark-and-sweep mechanism. Instead of doing a single `full` mutation, it performs a series of bursts. At the end of each burst, the Incremental Entity Provider performs the following operations, 1. Marks each received entity in the database. -2. Annotates each entity with `backstage/incremental-entity-provider: ` annotation. -3. Commits all of the entities with a `delta` mutation. +2. Commits all of the entities with a `delta` mutation. Incremental Entity Providers will wait a configurable interval before proceeding to the next burst. 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 a47cd8f58c..d64c1318ac 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts @@ -15,11 +15,7 @@ */ import type { DeferredEntity } from '@backstage/plugin-catalog-backend'; -import { - INCREMENTAL_ENTITY_PROVIDER_ANNOTATION, - IterationEngine, - IterationEngineOptions, -} from '../types'; +import { IterationEngine, IterationEngineOptions } from '../types'; import { IncrementalIngestionDatabaseManager } from '../database/IncrementalIngestionDatabaseManager'; import type { AbortSignal } from 'node-abort-controller'; import { performance } from 'perf_hooks'; @@ -271,8 +267,6 @@ export class IncrementalIngestionEngine implements IterationEngine { ...deferred.entity.metadata, annotations: { ...deferred.entity.metadata.annotations, - [INCREMENTAL_ENTITY_PROVIDER_ANNOTATION]: - this.options.provider.getProviderName(), }, }, }, diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts index c2603eea4e..ec46721a51 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts @@ -32,14 +32,6 @@ import type { DurationObjectUnits } from 'luxon'; import type { Logger } from 'winston'; import { IncrementalIngestionDatabaseManager } from './database/IncrementalIngestionDatabaseManager'; -/** - * Entity annotation containing the incremental entity provider. - * - * @public - */ -export const INCREMENTAL_ENTITY_PROVIDER_ANNOTATION = - 'backstage.io/incremental-provider-name'; - /** * Ingest entities into the catalog in bite-sized chunks. * From f34afe069b64acb3bfce1ef773833d0dbd3414a5 Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Tue, 29 Nov 2022 15:33:58 -0800 Subject: [PATCH 08/14] Remove stray annotation reference Signed-off-by: Damon Kaswell --- .../catalog-backend-module-incremental-ingestion/src/index.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/index.ts b/plugins/catalog-backend-module-incremental-ingestion/src/index.ts index 84d5fcf741..976c91a1ce 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/index.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/index.ts @@ -23,7 +23,6 @@ export * from './module'; export * from './service'; export { - INCREMENTAL_ENTITY_PROVIDER_ANNOTATION, type EntityIteratorResult, type IncrementalEntityProvider, type IncrementalEntityProviderOptions, From d2664469a6502fbc790628529e5f0f6c94933e1a Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Tue, 29 Nov 2022 15:58:19 -0800 Subject: [PATCH 09/14] Update api-report.md Signed-off-by: Damon Kaswell --- .../api-report.md | 4 ---- 1 file changed, 4 deletions(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/api-report.md b/plugins/catalog-backend-module-incremental-ingestion/api-report.md index 49eb143a4d..5f77490091 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/api-report.md +++ b/plugins/catalog-backend-module-incremental-ingestion/api-report.md @@ -30,10 +30,6 @@ export type EntityIteratorResult = cursor?: T; }; -// @public -export const INCREMENTAL_ENTITY_PROVIDER_ANNOTATION = - 'backstage.io/incremental-provider-name'; - // @public (undocumented) export class IncrementalCatalogBuilder { // (undocumented) From d99fd04856df4c5e16b359639c2111504e2127e3 Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Wed, 30 Nov 2022 08:22:23 -0800 Subject: [PATCH 10/14] Use refs for removed entities Signed-off-by: Damon Kaswell --- .../IncrementalIngestionDatabaseManager.ts | 15 +++------------ .../src/engine/IncrementalIngestionEngine.ts | 2 +- 2 files changed, 4 insertions(+), 13 deletions(-) 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 5cd8fce0be..c9299ec7d9 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts @@ -16,7 +16,7 @@ import { Knex } from 'knex'; import type { DeferredEntity } from '@backstage/plugin-catalog-backend'; -import { parseEntityRef, stringifyEntityRef } from '@backstage/catalog-model'; +import { stringifyEntityRef } from '@backstage/catalog-model'; import { Duration } from 'luxon'; import { v4 } from 'uuid'; import { @@ -312,17 +312,8 @@ export class IncrementalIngestionDatabaseManager { .join('ingestions', 'ingestions.id', 'ingestion_marks.ingestion_id') .where('ingestions.id', previousIngestion.id); - const removed: DeferredEntity[] = rows.map(e => { - const parsed = parseEntityRef(e.ref); - const entity = { - apiVersion: 'backstage.io/v1alpha1', - kind: parsed.kind, - metadata: { - name: parsed.name, - namespace: parsed.namespace, - }, - }; - return { entity }; + const removed: { entityRef: string }[] = rows.map(e => { + return { entityRef: e.ref }; }); const total = rows.length ?? 0; return { removed, total }; 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 5d82a29b5f..65deb92ad1 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts @@ -271,7 +271,7 @@ export class IncrementalIngestionEngine implements IterationEngine { }, })) ?? []; - const removed: DeferredEntity[] = []; + const removed: { entityRef: string }[] = []; if (done) { this.options.logger.info( From 1520967c504c46da4c70328632a7c0f730456656 Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Wed, 30 Nov 2022 15:16:59 -0800 Subject: [PATCH 11/14] 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; } From e3576b80ec3bd6447597a45ad5fdce9a40c2efbd Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Wed, 30 Nov 2022 15:45:21 -0800 Subject: [PATCH 12/14] Use better names Signed-off-by: Damon Kaswell --- .../api-report.md | 4 ++-- .../src/engine/IncrementalIngestionEngine.ts | 6 +++--- .../src/types.ts | 16 ++++++++-------- 3 files changed, 13 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 aef16e1533..387e505c33 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/api-report.md +++ b/plugins/catalog-backend-module-incremental-ingestion/api-report.md @@ -62,8 +62,8 @@ export interface IncrementalEntityProviderOptions { backoff?: DurationObjectUnits[]; burstInterval: DurationObjectUnits; burstLength: DurationObjectUnits; - rejectEmptyEntityCollections?: boolean; - removalThreshold?: number; + rejectEmptySourceCollections?: boolean; + rejectRemovalsAbovePercentage?: number; restLength: DurationObjectUnits; } 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 8df74b8891..e2000c609e 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts @@ -283,7 +283,7 @@ export class IncrementalIngestionEngine implements IterationEngine { const { total } = result; let doRemoval = true; - if (this.options.rejectEmptyEntityCollections) { + if (this.options.rejectEmptySourceCollections) { if (total === 0) { this.options.logger.error( `incremental-engine: Ingestion '${id}': Rejecting empty entity collection!`, @@ -292,12 +292,12 @@ export class IncrementalIngestionEngine implements IterationEngine { } } - if (this.options.removalThreshold) { + if (this.options.rejectRemovalsAbovePercentage) { // 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) { + if (percentRemoved <= this.options.rejectRemovalsAbovePercentage) { this.options.logger.info( `incremental-engine: Ingestion '${id}': Removing ${result.removed.length} entities that have no matching assets`, ); diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts index 657073b9fb..51a8738500 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts @@ -129,15 +129,15 @@ export interface IncrementalEntityProviderOptions { * Backstage removing all associated entities. To avoid that, set * a percentage of entities past which removal will be disallowed. */ - removalThreshold?: number; + rejectRemovalsAbovePercentage?: 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. + * Similar to the rejectRemovalsAbovePercentage, 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; + rejectEmptySourceCollections?: boolean; } /** @public */ @@ -162,6 +162,6 @@ export interface IterationEngineOptions { restLength: DurationObjectUnits; ready: Promise; backoff?: IncrementalEntityProviderOptions['backoff']; - removalThreshold?: number; - rejectEmptyEntityCollections?: boolean; + rejectRemovalsAbovePercentage?: number; + rejectEmptySourceCollections?: boolean; } From a579b64bc6297dbe6de737063c53b47a1a6c83e8 Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Fri, 2 Dec 2022 09:03:59 -0800 Subject: [PATCH 13/14] Speed improvement and fix typo Signed-off-by: Damon Kaswell --- .../IncrementalIngestionDatabaseManager.ts | 20 ++++++++++--------- 1 file changed, 11 insertions(+), 9 deletions(-) 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 5b1618a870..9b7afd6d9b 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts @@ -298,7 +298,7 @@ export class IncrementalIngestionDatabaseManager { * @param ingestionId - string * @returns All entities to remove for this burst. */ - async computeRemoved(provider: string, providerId: string) { + async computeRemoved(provider: string, ingestionId: string) { const previousIngestion = await this.getPreviousIngestionRecord(provider); return await this.client.transaction(async tx => { const count = await tx('ingestion_mark_entities') @@ -309,7 +309,7 @@ export class IncrementalIngestionDatabaseManager { 'ingestion_mark_entities.ingestion_mark_id', ) .join('ingestions', 'ingestions.id', 'ingestion_marks.ingestion_id') - .where('ingestions.id', providerId); + .where('ingestions.id', ingestionId); const total = count.reduce((acc, cur) => acc + (cur.total as number), 0); @@ -563,17 +563,19 @@ export class IncrementalIngestionDatabaseManager { const refs = entities.map(e => stringifyEntityRef(e.entity)); await this.client.transaction(async tx => { - const existingRefs = ( - await tx<{ ref: string }>('ingestion_mark_entities') - .select('ref') - .whereIn('ref', refs) - ).map(e => e.ref); + const existingRefs = new Set( + ...( + await tx<{ ref: string }>('ingestion_mark_entities') + .select('ref') + .whereIn('ref', refs) + ).map(e => e.ref), + ); - const newRefs = refs.filter(e => !existingRefs.includes(e)); + const newRefs = refs.filter(e => !existingRefs.has(e)); await tx('ingestion_mark_entities') .update('ingestion_mark_id', markId) - .whereIn('ref', existingRefs); + .whereIn('ref', Array.from(existingRefs)); await tx('ingestion_mark_entities').insert( newRefs.map(ref => ({ From f278c1bd79a76e8a12a717f57708608d184c5e62 Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Mon, 5 Dec 2022 08:06:33 -0800 Subject: [PATCH 14/14] Fix set bug and use more efficient vars Signed-off-by: Damon Kaswell --- .../IncrementalIngestionDatabaseManager.ts | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) 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 9b7afd6d9b..9eb52c4558 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts @@ -563,19 +563,19 @@ export class IncrementalIngestionDatabaseManager { const refs = entities.map(e => stringifyEntityRef(e.entity)); await this.client.transaction(async tx => { - const existingRefs = new Set( - ...( - await tx<{ ref: string }>('ingestion_mark_entities') - .select('ref') - .whereIn('ref', refs) - ).map(e => e.ref), - ); + const existingRefsArray = ( + await tx<{ ref: string }>('ingestion_mark_entities') + .select('ref') + .whereIn('ref', refs) + ).map(e => e.ref); - const newRefs = refs.filter(e => !existingRefs.has(e)); + const existingRefsSet = new Set(existingRefsArray); + + const newRefs = refs.filter(e => !existingRefsSet.has(e)); await tx('ingestion_mark_entities') .update('ingestion_mark_id', markId) - .whereIn('ref', Array.from(existingRefs)); + .whereIn('ref', existingRefsArray); await tx('ingestion_mark_entities').insert( newRefs.map(ref => ({