paulononato/backstage
1
0
Fork 0
Code Issues Pull Requests Actions 49 Packages Projects Releases Wiki Activity

Merge pull request #14904 from dekoding/dekoding/more-resilient-incremental-providers

Browse Source 
Make incremental entity providers more resilient to failed data sources
This commit is contained in:
Fredrik Adelöw
2022-12-06 10:29:15 +01:00
committed by GitHub
parent 325ebb0b34 892f71ef56
commit b722bd479f
7 changed files with 137 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/sharp-shoes-bathe.md
+5
View File
@@ -0,0 +1,5 @@
---
'@backstage/plugin-catalog-backend-module-incremental-ingestion': patch
---
Make incremental providers more resilient to failures
plugins/catalog-backend-module-incremental-ingestion/README.md
+2 -3
View File
@@ -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: <entity-provider-id>` 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.
plugins/catalog-backend-module-incremental-ingestion/api-report.md
+2 -4
View File
@@ -30,10 +30,6 @@ export type EntityIteratorResult<T> =
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;
}
plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts
+63 -33
View File
@@ -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<IngestionRecord>('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,
})),
);
});
plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts
+47 -13
View File
@@ -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({
plugins/catalog-backend-module-incremental-ingestion/src/index.ts
-1
View File
@@ -23,7 +23,6 @@
export * from './module';
export * from './service';
export {
INCREMENTAL_ENTITY_PROVIDER_ANNOTATION,
type EntityIteratorResult,
type IncrementalEntityProvider,
type IncrementalEntityProviderOptions,
plugins/catalog-backend-module-incremental-ingestion/src/types.ts
+18 -8
View File
@@ -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<void>;
backoff?: IncrementalEntityProviderOptions['backoff'];
rejectRemovalsAbovePercentage?: number;
rejectEmptySourceCollections?: boolean;
}