diff --git a/.changeset/sharp-shoes-bathe.md b/.changeset/sharp-shoes-bathe.md new file mode 100644 index 0000000000..8d3c0dfddb --- /dev/null +++ b/.changeset/sharp-shoes-bathe.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend-module-incremental-ingestion': patch +--- + +Make incremental providers more resilient to failures 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/api-report.md b/plugins/catalog-backend-module-incremental-ingestion/api-report.md index 49eb143a4d..387e505c33 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) @@ -66,6 +62,8 @@ export interface IncrementalEntityProviderOptions { backoff?: DurationObjectUnits[]; burstInterval: DurationObjectUnits; burstLength: DurationObjectUnits; + rejectEmptySourceCollections?: boolean; + rejectRemovalsAbovePercentage?: 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 3c91b24e2c..9eb52c4558 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts @@ -19,7 +19,6 @@ import type { DeferredEntity } from '@backstage/plugin-catalog-backend'; import { stringifyEntityRef } from '@backstage/catalog-model'; import { Duration } from 'luxon'; import { v4 } from 'uuid'; -import { INCREMENTAL_ENTITY_PROVIDER_ANNOTATION } from '../types'; import { IngestionRecord, IngestionRecordUpdate, @@ -113,6 +112,20 @@ 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 => { + return await tx('ingestions') + .where('provider_name', provider) + .andWhereNot('completion_ticket', 'open') + .first(); + }); + } + /** * 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,39 +299,40 @@ export class IncrementalIngestionDatabaseManager { * @returns All entities to remove for this burst. */ async computeRemoved(provider: string, ingestionId: string) { + const previousIngestion = await this.getPreviousIngestionRecord(provider); return await this.client.transaction(async tx => { - 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'), - ) + const count = await tx('ingestion_mark_entities') + .count({ total: 'ingestion_mark_entities.ref' }) .join( - 'refresh_state', - 'refresh_state.entity_id', - 'final_entities.entity_id', + 'ingestion_marks', + 'ingestion_marks.id', + 'ingestion_mark_entities.ingestion_mark_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 removed.map(entity => { - return { entity: JSON.parse(entity.entity) }; - }); + .join('ingestions', 'ingestions.id', 'ingestion_marks.ingestion_id') + .where('ingestions.id', ingestionId); + + const total = count.reduce((acc, cur) => acc + (cur.total as number), 0); + + const removed: { entityRef: string }[] = []; + if (previousIngestion) { + const stale: { ref: string }[] = await tx('ingestion_mark_entities') + .select('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', previousIngestion.id); + + removed.push( + ...stale.map(e => { + return { entityRef: e.ref }; + }), + ); + } + + return { total, removed }; }); } @@ -546,12 +560,28 @@ 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 existingRefsArray = ( + await tx<{ ref: string }>('ingestion_mark_entities') + .select('ref') + .whereIn('ref', refs) + ).map(e => e.ref); + + 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', existingRefsArray); + 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 86c58b0082..e2000c609e 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 { performance } from 'perf_hooks'; import { Duration, DurationObjectUnits } from 'luxon'; @@ -268,22 +264,60 @@ export class IncrementalIngestionEngine implements IterationEngine { ...deferred.entity.metadata, annotations: { ...deferred.entity.metadata.annotations, - [INCREMENTAL_ENTITY_PROVIDER_ANNOTATION]: - this.options.provider.getProviderName(), }, }, }, })) ?? []; - const removed: DeferredEntity[] = []; + const removed: { entityRef: string }[] = []; 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; + + let doRemoval = true; + if (this.options.rejectEmptySourceCollections) { + if (total === 0) { + this.options.logger.error( + `incremental-engine: Ingestion '${id}': Rejecting empty entity collection!`, + ); + doRemoval = false; + } + } + + 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.rejectRemovalsAbovePercentage) { + 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); + } } await this.options.connection.applyMutation({ 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, diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts index c2603eea4e..51a8738500 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. * @@ -130,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. + */ + rejectRemovalsAbovePercentage?: number; + + /** + * 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. + */ + rejectEmptySourceCollections?: boolean; } /** @public */ @@ -154,4 +162,6 @@ export interface IterationEngineOptions { restLength: DurationObjectUnits; ready: Promise; backoff?: IncrementalEntityProviderOptions['backoff']; + rejectRemovalsAbovePercentage?: number; + rejectEmptySourceCollections?: boolean; }