diff --git a/plugins/catalog-backend/api-report.md b/plugins/catalog-backend/api-report.md index ad898d2e22..8af2643dc0 100644 --- a/plugins/catalog-backend/api-report.md +++ b/plugins/catalog-backend/api-report.md @@ -145,10 +145,13 @@ export class CatalogBuilder { processingInterval: ProcessingIntervalFunction, ): CatalogBuilder; setProcessingIntervalSeconds(seconds: number): CatalogBuilder; - // @alpha (undocumented) - subscribe( - catalogProcessingErrorListeners: CatalogProcessingErrorListener[], - ): void; + // (undocumented) + subscribe(options: { + onProcessingError: (event: { + unprocessedEntity: Entity; + errors: Error[]; + }) => Promise | void; + }): void; } // @alpha @@ -212,16 +215,6 @@ export interface CatalogProcessingEngine { stop(): Promise; } -// @alpha -export interface CatalogProcessingErrorListener { - // (undocumented) - onError( - unprocessedEntity: Entity, - result: EntityProcessingResult, - resultHash: String, - ): Promise; -} - // @public (undocumented) export type CatalogProcessor = { getProcessorName(): string; diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts index 69295a0662..b632b48825 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts @@ -14,8 +14,8 @@ * limitations under the License. */ -import { stringifyEntityRef } from '@backstage/catalog-model'; -import { assertError, serializeError } from '@backstage/errors'; +import { Entity, stringifyEntityRef } from '@backstage/catalog-model'; +import { assertError, serializeError, stringifyError } from '@backstage/errors'; import { Hash } from 'crypto'; import stableStringify from 'fast-json-stable-stringify'; import { Logger } from 'winston'; @@ -23,7 +23,6 @@ import { ProcessingDatabase, RefreshStateItem } from '../database/types'; import { createCounterMetric, createSummaryMetric } from '../util/metrics'; import { CatalogProcessingEngine, - CatalogProcessingErrorListener, CatalogProcessingOrchestrator, EntityProcessingResult, } from './types'; @@ -43,7 +42,10 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { private readonly stitcher: Stitcher, private readonly createHash: () => Hash, private readonly pollingIntervalMs: number = 1000, - private readonly catalogProcessingErrorListener?: CatalogProcessingErrorListener, + private readonly onProcessingError?: (event: { + unprocessedEntity: Entity; + errors: Error[]; + }) => Promise | void, ) {} async start() { @@ -158,11 +160,20 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { // the outside. if (!result.ok) { // notify the error listener if the entity can not be processed. - this.catalogProcessingErrorListener?.onError( - unprocessedEntity, - result, - resultHash, - ); + Promise.resolve(undefined) + .then(() => + this.onProcessingError?.({ + unprocessedEntity, + errors: result.errors, + }), + ) + .catch(error => { + this.logger.debug( + `Processing error listener threw an exception, ${stringifyError( + error, + )}`, + ); + }); await this.processingDatabase.transaction(async tx => { await this.processingDatabase.updateProcessedEntityErrors(tx, { diff --git a/plugins/catalog-backend/src/processing/index.ts b/plugins/catalog-backend/src/processing/index.ts index c5e139c0ad..d96e7b5b9b 100644 --- a/plugins/catalog-backend/src/processing/index.ts +++ b/plugins/catalog-backend/src/processing/index.ts @@ -18,7 +18,6 @@ export type { CatalogProcessingEngine, EntityProcessingResult, DeferredEntity, - CatalogProcessingErrorListener, } from './types'; export { createRandomProcessingInterval } from './refresh'; diff --git a/plugins/catalog-backend/src/processing/types.ts b/plugins/catalog-backend/src/processing/types.ts index e6c58cf36f..b178522c14 100644 --- a/plugins/catalog-backend/src/processing/types.ts +++ b/plugins/catalog-backend/src/processing/types.ts @@ -67,16 +67,3 @@ export interface CatalogProcessingEngine { start(): Promise; stop(): Promise; } - -/** - * An error listener for catalog processing engine. It can be used to listen and track entity errors. - * - * @alpha - */ -export interface CatalogProcessingErrorListener { - onError( - unprocessedEntity: Entity, - result: EntityProcessingResult, - resultHash: String, - ): Promise; -} diff --git a/plugins/catalog-backend/src/service/CatalogBuilder.ts b/plugins/catalog-backend/src/service/CatalogBuilder.ts index be427998fc..390dc817d1 100644 --- a/plugins/catalog-backend/src/service/CatalogBuilder.ts +++ b/plugins/catalog-backend/src/service/CatalogBuilder.ts @@ -17,6 +17,7 @@ import { PluginDatabaseManager, UrlReader } from '@backstage/backend-common'; import { DefaultNamespaceEntityPolicy, + Entity, EntityPolicies, EntityPolicy, FieldFormatEntityPolicy, @@ -56,10 +57,7 @@ import { } from '../modules/core/PlaceholderProcessor'; import { defaultEntityDataParser } from '../modules/util/parse'; import { LocationAnalyzer } from '../ingestion/types'; -import { - CatalogProcessingEngine, - CatalogProcessingErrorListener, -} from '../processing'; +import { CatalogProcessingEngine } from '../processing'; import { DefaultProcessingDatabase } from '../database/DefaultProcessingDatabase'; import { applyDatabaseMigrations } from '../database/migrations'; import { DefaultCatalogProcessingEngine } from '../processing/DefaultCatalogProcessingEngine'; @@ -136,7 +134,10 @@ export class CatalogBuilder { private processors: CatalogProcessor[]; private processorsReplace: boolean; private parser: CatalogProcessorParser | undefined; - private catalogProcessingErrorListeners?: CatalogProcessingErrorListener[]; + private onProcessingError?: (event: { + unprocessedEntity: Entity; + errors: Error[]; + }) => Promise | void; private processingInterval: ProcessingIntervalFunction = createRandomProcessingInterval({ minSeconds: 100, @@ -452,12 +453,8 @@ export class CatalogBuilder { stitcher, () => createHash('sha1'), 1000, - { - onError: async (unprocessedEntity, result, resultHash) => { - this.catalogProcessingErrorListeners?.forEach(listener => - listener.onError(unprocessedEntity, result, resultHash), - ); - }, + event => { + this.onProcessingError?.(event); }, ); @@ -490,12 +487,13 @@ export class CatalogBuilder { }; } - /** - * @alpha - * @param catalogProcessingErrorListeners - a list of listeners to get notified if an error occurs while processing an entity - */ - subscribe(catalogProcessingErrorListeners: CatalogProcessingErrorListener[]) { - this.catalogProcessingErrorListeners = catalogProcessingErrorListeners; + subscribe(options: { + onProcessingError: (event: { + unprocessedEntity: Entity; + errors: Error[]; + }) => Promise | void; + }) { + this.onProcessingError = options.onProcessingError; } private buildEntityPolicy(): EntityPolicy {