diff --git a/.changeset/shaggy-spiders-notice.md b/.changeset/shaggy-spiders-notice.md new file mode 100644 index 0000000000..6b50c964e3 --- /dev/null +++ b/.changeset/shaggy-spiders-notice.md @@ -0,0 +1,6 @@ +--- +'@backstage/plugin-catalog-backend': patch +--- + +Provide a new listener `CatalogProcessingErrorListener` for the processing engines to notify the caller if an entity can not be processed. Processing engine +will collect the errors and passes the entity and the results to the given listeners. diff --git a/plugins/catalog-backend/api-report.md b/plugins/catalog-backend/api-report.md index cb8a5abb8a..3ac5094b31 100644 --- a/plugins/catalog-backend/api-report.md +++ b/plugins/catalog-backend/api-report.md @@ -15,6 +15,7 @@ import { DocumentCollatorFactory } from '@backstage/plugin-search-common'; import { Entity } from '@backstage/catalog-model'; import { EntityPolicy } from '@backstage/catalog-model'; import { GetEntitiesRequest } from '@backstage/catalog-client'; +import { JsonObject } from '@backstage/types'; import { JsonValue } from '@backstage/types'; import { LocationEntityV1alpha1 } from '@backstage/catalog-model'; import { Logger } from 'winston'; @@ -201,12 +202,24 @@ export type CatalogPermissionRule = // @public (undocumented) export interface CatalogProcessingEngine { + // (undocumented) + addErrorListener?(errorListener: CatalogProcessingErrorListener): void; // (undocumented) start(): Promise; // (undocumented) stop(): Promise; } +// @public +export interface CatalogProcessingErrorListener { + // (undocumented) + onError( + unprocessedEntity: Entity, + result: EntityProcessingResult, + resultHash: String, + ): Promise; +} + // @public (undocumented) export type CatalogProcessor = { getProcessorName(): string; @@ -412,6 +425,21 @@ export type EntityFilter = } | EntitiesSearchFilter; +// @public +export type EntityProcessingResult = + | { + ok: true; + state: JsonObject; + completedEntity: Entity; + deferredEntities: DeferredEntity[]; + relations: EntityRelationSpec[]; + errors: Error[]; + } + | { + ok: false; + errors: Error[]; + }; + // @public export interface EntityProvider { connect(connection: EntityProviderConnection): Promise; diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts index b7ef5c9344..7ebb55273b 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts @@ -23,9 +23,10 @@ import { ProcessingDatabase, RefreshStateItem } from '../database/types'; import { createCounterMetric, createSummaryMetric } from '../util/metrics'; import { CatalogProcessingEngine, + CatalogProcessingErrorListener, CatalogProcessingOrchestrator, EntityProcessingResult, -} from '../processing/types'; +} from './types'; import { Stitcher } from '../stitching/Stitcher'; import { startTaskPipeline } from './TaskPipeline'; @@ -33,6 +34,7 @@ const CACHE_TTL = 5; export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { private readonly tracker = progressTracker(); + private readonly errorListeners: CatalogProcessingErrorListener[] = []; private stopFunc?: () => void; constructor( @@ -122,6 +124,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { ); let hashBuilder = this.createHash().update(errorsString); + if (result.ok) { const { entityRefs: parents } = await this.processingDatabase.transaction(tx => @@ -154,6 +157,11 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { // just store the errors and trigger a stich so that they become visible to // the outside. if (!result.ok) { + // notify the error listeners if the entity can not be processed. + this.errorListeners.forEach(listener => + listener.onError(unprocessedEntity, result, resultHash), + ); + await this.processingDatabase.transaction(async tx => { await this.processingDatabase.updateProcessedEntityErrors(tx, { id, @@ -223,6 +231,10 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { this.stopFunc = undefined; } } + + addErrorListener(errorListener: CatalogProcessingErrorListener) { + this.errorListeners.push(errorListener); + } } // Helps wrap the timing and logging behaviors diff --git a/plugins/catalog-backend/src/processing/index.ts b/plugins/catalog-backend/src/processing/index.ts index b392c3aa76..c5e139c0ad 100644 --- a/plugins/catalog-backend/src/processing/index.ts +++ b/plugins/catalog-backend/src/processing/index.ts @@ -14,7 +14,12 @@ * limitations under the License. */ -export type { CatalogProcessingEngine, DeferredEntity } from './types'; +export type { + CatalogProcessingEngine, + EntityProcessingResult, + DeferredEntity, + CatalogProcessingErrorListener, +} from './types'; export { createRandomProcessingInterval } from './refresh'; export type { ProcessingIntervalFunction } from './refresh'; diff --git a/plugins/catalog-backend/src/processing/types.ts b/plugins/catalog-backend/src/processing/types.ts index b178522c14..54118f8422 100644 --- a/plugins/catalog-backend/src/processing/types.ts +++ b/plugins/catalog-backend/src/processing/types.ts @@ -66,4 +66,18 @@ export type DeferredEntity = { export interface CatalogProcessingEngine { start(): Promise; stop(): Promise; + addErrorListener?(errorListener: CatalogProcessingErrorListener): void; +} + +/** + * An error listener for catalog processing engine. It can be used to listen and track entity errors. + * + * @public + */ +export interface CatalogProcessingErrorListener { + onError( + unprocessedEntity: Entity, + result: EntityProcessingResult, + resultHash: String, + ): Promise; }