diff --git a/plugins/catalog-backend/migrations/20210813143113_add_refresh_state_hash.js b/plugins/catalog-backend/migrations/20210813143113_add_refresh_state_hash.js index 5ac9a95689..29a102e876 100644 --- a/plugins/catalog-backend/migrations/20210813143113_add_refresh_state_hash.js +++ b/plugins/catalog-backend/migrations/20210813143113_add_refresh_state_hash.js @@ -22,7 +22,7 @@ exports.up = async function up(knex) { await knex.schema.alterTable('refresh_state', table => { table - .text('hash') + .text('result_hash') .nullable() .comment( 'A hash of the processed contents, used to avoid duplicate work', @@ -35,6 +35,6 @@ exports.up = async function up(knex) { */ exports.down = async function down(knex) { await knex.schema.alterTable('refresh_state', table => { - table.dropColumn('hash'); + table.dropColumn('result_hash'); }); }; diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts index 794f2e6d0b..898ef08f6c 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts @@ -15,6 +15,7 @@ */ import { getVoidLogger } from '@backstage/backend-common'; +import { Hash } from 'crypto'; import waitForExpect from 'wait-for-expect'; import { DefaultProcessingDatabase } from './database/DefaultProcessingDatabase'; import { DefaultCatalogProcessingEngine } from './DefaultCatalogProcessingEngine'; @@ -33,6 +34,10 @@ describe('DefaultCatalogProcessingEngine', () => { const stitcher = { stitch: jest.fn(), } as unknown as jest.Mocked; + const hash = { + update: () => hash, + digest: jest.fn(), + } as unknown as jest.Mocked; beforeEach(() => { jest.resetAllMocks(); @@ -57,6 +62,7 @@ describe('DefaultCatalogProcessingEngine', () => { db, orchestrator, stitcher, + () => hash, ); db.transaction.mockImplementation(cb => cb((() => {}) as any)); @@ -76,7 +82,7 @@ describe('DefaultCatalogProcessingEngine', () => { kind: 'Location', metadata: { name: 'test' }, }, - hash: '', + resultHash: '', state: new Map(), nextUpdateAt: '', lastDiscoveryAt: '', @@ -118,6 +124,7 @@ describe('DefaultCatalogProcessingEngine', () => { db, orchestrator, stitcher, + () => hash, ); db.transaction.mockImplementation(cb => cb((() => {}) as any)); @@ -138,7 +145,7 @@ describe('DefaultCatalogProcessingEngine', () => { kind: 'Location', metadata: { name: 'test' }, }, - hash: '', + resultHash: '', state: new Map(), nextUpdateAt: '', lastDiscoveryAt: '', @@ -160,4 +167,71 @@ describe('DefaultCatalogProcessingEngine', () => { }); await engine.stop(); }); + + it('runs fully when hash mismatches, early-outs when hash matches', async () => { + const entity = { + apiVersion: '1', + kind: 'Location', + metadata: { name: 'test' }, + }; + + const refreshState = { + id: '', + entityRef: '', + unprocessedEntity: entity, + resultHash: 'the matching hash', + state: new Map(), + nextUpdateAt: '', + lastDiscoveryAt: '', + }; + + hash.digest.mockReturnValue('the matching hash'); + + orchestrator.process.mockResolvedValue({ + ok: true, + completedEntity: entity, + relations: [], + errors: [], + deferredEntities: [], + state: new Map(), + }); + + const engine = new DefaultCatalogProcessingEngine( + getVoidLogger(), + [], + db, + orchestrator, + stitcher, + () => hash, + ); + + db.transaction.mockImplementation(cb => cb((() => {}) as any)); + + db.getProcessableEntities + .mockResolvedValueOnce({ + items: [{ ...refreshState, resultHash: 'NOT RIGHT' }], + }) + .mockResolvedValue({ items: [] }); + + await engine.start(); + + await waitForExpect(() => { + expect(orchestrator.process).toBeCalledTimes(1); + expect(hash.digest).toBeCalledTimes(1); + expect(db.updateProcessedEntity).toBeCalledTimes(1); + }); + + db.getProcessableEntities + .mockReset() + .mockResolvedValueOnce({ items: [refreshState] }) + .mockResolvedValue({ items: [] }); + + await waitForExpect(() => { + expect(orchestrator.process).toBeCalledTimes(2); + expect(hash.digest).toBeCalledTimes(2); + expect(db.updateProcessedEntity).toBeCalledTimes(1); + }); + + await engine.stop(); + }); }); diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index 8e4a4ddc0c..6468fa5187 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -20,7 +20,7 @@ import { stringifyEntityRef, } from '@backstage/catalog-model'; import { serializeError } from '@backstage/errors'; -import { createHash } from 'crypto'; +import { Hash } from 'crypto'; import stableStringify from 'fast-json-stable-stringify'; import { Logger } from 'winston'; import { ProcessingDatabase, RefreshStateItem } from './database/types'; @@ -91,6 +91,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { private readonly processingDatabase: ProcessingDatabase, private readonly orchestrator: CatalogProcessingOrchestrator, private readonly stitcher: Stitcher, + private readonly createHash: () => Hash, ) {} async start() { @@ -133,7 +134,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { unprocessedEntity, entityRef, locationKey, - hash: previousHash, + resultHash: previousResultHash, } = item; const result = await this.orchestrator.process({ entity: unprocessedEntity, @@ -151,17 +152,17 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { result.errors.map(e => serializeError(e)), ); - let hashBuilder = createHash('sha1').update(errorsString); + let hashBuilder = this.createHash().update(errorsString); if (result.ok) { hashBuilder = hashBuilder .update(stableStringify({ ...result.completedEntity })) .update(stableStringify([...result.deferredEntities])) .update(stableStringify([...result.relations])) - .update(stableStringify({ ...result.state })); + .update(stableStringify(Object.fromEntries(result.state))); } - const hash = hashBuilder.digest('hex'); - if (hash === previousHash) { + const resultHash = hashBuilder.digest('hex'); + if (resultHash === previousResultHash) { // If nothing changed in our produced outputs, we cannot have any // significant effect on our surroundings; therefore, we just abort // without any updates / stitching. @@ -180,7 +181,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { await this.processingDatabase.updateProcessedEntityErrors(tx, { id, errors: errorsString, - hash, + resultHash, }); }); await this.stitcher.stitch( @@ -194,7 +195,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { await this.processingDatabase.updateProcessedEntity(tx, { id, processedEntity: result.completedEntity, - hash, + resultHash, state: result.state, errors: errorsString, relations: result.relations, diff --git a/plugins/catalog-backend/src/next/NextCatalogBuilder.ts b/plugins/catalog-backend/src/next/NextCatalogBuilder.ts index a525dacd51..fb12077508 100644 --- a/plugins/catalog-backend/src/next/NextCatalogBuilder.ts +++ b/plugins/catalog-backend/src/next/NextCatalogBuilder.ts @@ -31,6 +31,7 @@ import { } from '@backstage/catalog-model'; import { Config } from '@backstage/config'; import { ScmIntegrations } from '@backstage/integration'; +import { createHash } from 'crypto'; import lodash from 'lodash'; import { Logger } from 'winston'; import { @@ -310,6 +311,7 @@ export class NextCatalogBuilder { processingDatabase, orchestrator, stitcher, + () => createHash('sha1'), ); const locationsCatalog = new DatabaseLocationsCatalog(db); diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts index 8b59e5b5da..5103c25dbc 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts @@ -19,8 +19,8 @@ import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils'; import { Entity, stringifyEntityRef } from '@backstage/catalog-model'; import { JsonObject } from '@backstage/config'; import { Knex } from 'knex'; -import { Logger } from 'winston'; import * as uuid from 'uuid'; +import { Logger } from 'winston'; import { DatabaseManager } from './DatabaseManager'; import { DefaultProcessingDatabase } from './DefaultProcessingDatabase'; import { @@ -206,7 +206,7 @@ describe('Default Processing Database', () => { db.updateProcessedEntity(tx, { id, processedEntity, - hash: '', + resultHash: '', state: new Map(), relations: [], deferredEntities: [], @@ -225,7 +225,7 @@ describe('Default Processing Database', () => { const options = { id, processedEntity, - hash: '', + resultHash: '', state: new Map(), relations: [], deferredEntities: [], @@ -254,7 +254,7 @@ describe('Default Processing Database', () => { expect( db.updateProcessedEntity(tx, { ...options, - hash: '', + resultHash: '', locationKey: 'fail', }), ).rejects.toThrow( @@ -286,7 +286,7 @@ describe('Default Processing Database', () => { db.updateProcessedEntity(tx, { id, processedEntity, - hash: '', + resultHash: '', state, relations: [], deferredEntities: [], @@ -302,7 +302,9 @@ describe('Default Processing Database', () => { expect(entities[0].processed_entity).toEqual( JSON.stringify(processedEntity), ); - expect(entities[0].cache).toEqual(JSON.stringify(state)); + expect(entities[0].cache).toEqual( + JSON.stringify(Object.fromEntries(state)), + ); expect(entities[0].errors).toEqual("['something broke']"); expect(entities[0].location_key).toEqual('key'); }, @@ -343,7 +345,7 @@ describe('Default Processing Database', () => { db.updateProcessedEntity(tx, { id, processedEntity, - hash: '', + resultHash: '', state: new Map(), relations: relations, deferredEntities: [], @@ -395,7 +397,7 @@ describe('Default Processing Database', () => { db.updateProcessedEntity(tx, { id, processedEntity, - hash: '', + resultHash: '', 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 bb5f50c7a0..177bf979ab 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -60,7 +60,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { const { id, processedEntity, - hash, + resultHash, state, errors, relations, @@ -70,8 +70,8 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { const refreshResult = await tx('refresh_state') .update({ processed_entity: JSON.stringify(processedEntity), - hash, - cache: JSON.stringify(state), + result_hash: resultHash, + cache: JSON.stringify(Object.fromEntries(state || [])), errors, location_key: locationKey, }) @@ -124,12 +124,12 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { options: UpdateProcessedEntityOptions, ): Promise { const tx = txOpaque as Knex.Transaction; - const { id, errors, hash } = options; + const { id, errors, resultHash } = options; await tx('refresh_state') .update({ errors, - hash, + result_hash: resultHash, }) .where('entity_id', id); } @@ -495,7 +495,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { processedEntity: i.processed_entity ? (JSON.parse(i.processed_entity) as Entity) : undefined, - hash: i.hash || '', + resultHash: i.result_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 594b7a3ade..d735607554 100644 --- a/plugins/catalog-backend/src/next/database/tables.ts +++ b/plugins/catalog-backend/src/next/database/tables.ts @@ -25,7 +25,7 @@ export type DbRefreshStateRow = { entity_ref: string; unprocessed_entity: string; processed_entity?: string; - hash?: string; + result_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 c559895d27..c93b90084d 100644 --- a/plugins/catalog-backend/src/next/database/types.ts +++ b/plugins/catalog-backend/src/next/database/types.ts @@ -34,7 +34,7 @@ export type AddUnprocessedEntitiesResult = {}; export type UpdateProcessedEntityOptions = { id: string; processedEntity: Entity; - hash: string; + resultHash: string; state?: Map; errors?: string; relations: EntityRelationSpec[]; @@ -45,7 +45,7 @@ export type UpdateProcessedEntityOptions = { export type UpdateProcessedEntityErrorsOptions = { id: string; errors?: string; - hash: string; + resultHash: string; }; export type RefreshStateItem = { @@ -53,7 +53,7 @@ export type RefreshStateItem = { entityRef: string; unprocessedEntity: Entity; processedEntity?: Entity; - hash: string; + resultHash: string; nextUpdateAt: string; lastDiscoveryAt: string; // remove? state: Map;