diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index 0ae20c94b1..89bf3616ca 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -20,6 +20,8 @@ import { stringifyEntityRef, } from '@backstage/catalog-model'; import { serializeError } from '@backstage/errors'; +import { createHash } from 'crypto'; +import stableStringify from 'fast-json-stable-stringify'; import { Logger } from 'winston'; import { ProcessingDatabase, RefreshStateItem } from './database/types'; import { CatalogProcessingOrchestrator } from './processing/types'; @@ -125,7 +127,14 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { }, processTask: async item => { try { - const { id, state, unprocessedEntity, entityRef, locationKey } = item; + const { + id, + state, + unprocessedEntity, + entityRef, + locationKey, + hash: previousHash, + } = item; const result = await this.orchestrator.process({ entity: unprocessedEntity, state, @@ -142,6 +151,22 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { result.errors.map(e => serializeError(e)), ); + let hashBuilder = createHash('sha1').update(errorsString); + if (result.ok) { + hashBuilder = hashBuilder + .update(stableStringify({ ...result.completedEntity })) + .update(stableStringify([...result.deferredEntities])) + .update(stableStringify([...result.relations])) + .update(stableStringify({ ...result.state })); + } + + const hash = hashBuilder.digest('hex'); + if (hash === previousHash) { + console.log('skipping ', entityRef); + return; + } + console.log('going ahead with ', entityRef); + // If the result was marked as not OK, it signals that some part of the // processing pipeline threw an exception. This can happen both as part of // non-catastrophic things such as due to validation errors, as well as if @@ -154,6 +179,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { await this.processingDatabase.updateProcessedEntityErrors(tx, { id, errors: errorsString, + hash, }); }); await this.stitcher.stitch( @@ -167,6 +193,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { await this.processingDatabase.updateProcessedEntity(tx, { id, processedEntity: result.completedEntity, + hash, state: result.state, errors: errorsString, relations: result.relations, diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts index 6a0626afc4..8b59e5b5da 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts @@ -206,6 +206,7 @@ describe('Default Processing Database', () => { db.updateProcessedEntity(tx, { id, processedEntity, + hash: '', state: new Map(), relations: [], deferredEntities: [], @@ -224,6 +225,7 @@ describe('Default Processing Database', () => { const options = { id, processedEntity, + hash: '', state: new Map(), relations: [], deferredEntities: [], @@ -250,7 +252,11 @@ describe('Default Processing Database', () => { await db.transaction(tx => expect( - db.updateProcessedEntity(tx, { ...options, locationKey: 'fail' }), + db.updateProcessedEntity(tx, { + ...options, + hash: '', + locationKey: 'fail', + }), ).rejects.toThrow( `Conflicting write of processing result for ${id} with location key 'fail'`, ), @@ -280,6 +286,7 @@ describe('Default Processing Database', () => { db.updateProcessedEntity(tx, { id, processedEntity, + hash: '', state, relations: [], deferredEntities: [], @@ -336,6 +343,7 @@ describe('Default Processing Database', () => { db.updateProcessedEntity(tx, { id, processedEntity, + hash: '', state: new Map(), relations: relations, deferredEntities: [], @@ -387,6 +395,7 @@ describe('Default Processing Database', () => { db.updateProcessedEntity(tx, { id, processedEntity, + hash: '', state: new Map(), relations: [], deferredEntities, diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index 1f1ea4bf84..1511854a03 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -60,6 +60,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { const { id, processedEntity, + hash, state, errors, relations, @@ -69,6 +70,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { const refreshResult = await tx('refresh_state') .update({ processed_entity: JSON.stringify(processedEntity), + hash, cache: JSON.stringify(state), errors, location_key: locationKey, @@ -492,6 +494,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { processedEntity: i.processed_entity ? (JSON.parse(i.processed_entity) as Entity) : undefined, + hash: i.hash, nextUpdateAt: i.next_update_at, lastDiscoveryAt: i.last_discovery_at, state: i.cache diff --git a/plugins/catalog-backend/src/next/database/tables.ts b/plugins/catalog-backend/src/next/database/tables.ts index 395694131d..594b7a3ade 100644 --- a/plugins/catalog-backend/src/next/database/tables.ts +++ b/plugins/catalog-backend/src/next/database/tables.ts @@ -25,6 +25,7 @@ export type DbRefreshStateRow = { entity_ref: string; unprocessed_entity: string; processed_entity?: string; + hash?: string; cache?: string; next_update_at: string; last_discovery_at: string; // remove? diff --git a/plugins/catalog-backend/src/next/database/types.ts b/plugins/catalog-backend/src/next/database/types.ts index 78199900e4..c559895d27 100644 --- a/plugins/catalog-backend/src/next/database/types.ts +++ b/plugins/catalog-backend/src/next/database/types.ts @@ -34,6 +34,7 @@ export type AddUnprocessedEntitiesResult = {}; export type UpdateProcessedEntityOptions = { id: string; processedEntity: Entity; + hash: string; state?: Map; errors?: string; relations: EntityRelationSpec[]; @@ -44,6 +45,7 @@ export type UpdateProcessedEntityOptions = { export type UpdateProcessedEntityErrorsOptions = { id: string; errors?: string; + hash: string; }; export type RefreshStateItem = { @@ -51,6 +53,7 @@ export type RefreshStateItem = { entityRef: string; unprocessedEntity: Entity; processedEntity?: Entity; + hash: string; nextUpdateAt: string; lastDiscoveryAt: string; // remove? state: Map;