diff --git a/.changeset/cuddly-flowers-film.md b/.changeset/cuddly-flowers-film.md new file mode 100644 index 0000000000..290bb74db6 --- /dev/null +++ b/.changeset/cuddly-flowers-film.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend': patch +--- + +Avoid duplicate work by comparing previous processing rounds with the next diff --git a/plugins/catalog-backend/knexfile.js b/plugins/catalog-backend/knexfile.js new file mode 100644 index 0000000000..4c8be42673 --- /dev/null +++ b/plugins/catalog-backend/knexfile.js @@ -0,0 +1,26 @@ +/* + * Copyright 2021 The Backstage Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +// This file makes it possible to run "yarn knex migrate:make some_file_name" +// to assist in making new migrations +module.exports = { + client: 'sqlite3', + connection: ':memory:', + useNullAsDefault: true, + migrations: { + directory: './migrations', + }, +}; diff --git a/plugins/catalog-backend/migrations/20210813143113_add_refresh_state_hash.js b/plugins/catalog-backend/migrations/20210813143113_add_refresh_state_hash.js new file mode 100644 index 0000000000..29a102e876 --- /dev/null +++ b/plugins/catalog-backend/migrations/20210813143113_add_refresh_state_hash.js @@ -0,0 +1,40 @@ +/* + * Copyright 2021 The Backstage Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +// @ts-check + +/** + * @param {import('knex').Knex} knex + */ +exports.up = async function up(knex) { + await knex.schema.alterTable('refresh_state', table => { + table + .text('result_hash') + .nullable() + .comment( + 'A hash of the processed contents, used to avoid duplicate work', + ); + }); +}; + +/** + * @param {import('knex').Knex} knex + */ +exports.down = async function down(knex) { + await knex.schema.alterTable('refresh_state', table => { + 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 5f3bab63e7..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,6 +82,7 @@ describe('DefaultCatalogProcessingEngine', () => { kind: 'Location', metadata: { name: 'test' }, }, + resultHash: '', state: new Map(), nextUpdateAt: '', lastDiscoveryAt: '', @@ -117,6 +124,7 @@ describe('DefaultCatalogProcessingEngine', () => { db, orchestrator, stitcher, + () => hash, ); db.transaction.mockImplementation(cb => cb((() => {}) as any)); @@ -137,6 +145,7 @@ describe('DefaultCatalogProcessingEngine', () => { kind: 'Location', metadata: { name: 'test' }, }, + resultHash: '', state: new Map(), nextUpdateAt: '', lastDiscoveryAt: '', @@ -158,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 0ae20c94b1..6468fa5187 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 { Hash } from 'crypto'; +import stableStringify from 'fast-json-stable-stringify'; import { Logger } from 'winston'; import { ProcessingDatabase, RefreshStateItem } from './database/types'; import { CatalogProcessingOrchestrator } from './processing/types'; @@ -89,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() { @@ -125,7 +128,14 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { }, processTask: async item => { try { - const { id, state, unprocessedEntity, entityRef, locationKey } = item; + const { + id, + state, + unprocessedEntity, + entityRef, + locationKey, + resultHash: previousResultHash, + } = item; const result = await this.orchestrator.process({ entity: unprocessedEntity, state, @@ -142,6 +152,23 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { result.errors.map(e => serializeError(e)), ); + 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(Object.fromEntries(result.state))); + } + + 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. + return; + } + // 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 +181,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { await this.processingDatabase.updateProcessedEntityErrors(tx, { id, errors: errorsString, + resultHash, }); }); await this.stitcher.stitch( @@ -167,6 +195,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { await this.processingDatabase.updateProcessedEntity(tx, { id, processedEntity: result.completedEntity, + 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 6a0626afc4..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,6 +206,7 @@ describe('Default Processing Database', () => { db.updateProcessedEntity(tx, { id, processedEntity, + resultHash: '', state: new Map(), relations: [], deferredEntities: [], @@ -224,6 +225,7 @@ describe('Default Processing Database', () => { const options = { id, processedEntity, + resultHash: '', 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, + resultHash: '', + 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, + resultHash: '', state, relations: [], deferredEntities: [], @@ -295,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'); }, @@ -336,6 +345,7 @@ describe('Default Processing Database', () => { db.updateProcessedEntity(tx, { id, processedEntity, + resultHash: '', state: new Map(), relations: relations, deferredEntities: [], @@ -387,6 +397,7 @@ describe('Default Processing Database', () => { db.updateProcessedEntity(tx, { id, processedEntity, + 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 1f1ea4bf84..177bf979ab 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, + resultHash, state, errors, relations, @@ -69,7 +70,8 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { const refreshResult = await tx('refresh_state') .update({ processed_entity: JSON.stringify(processedEntity), - cache: JSON.stringify(state), + result_hash: resultHash, + cache: JSON.stringify(Object.fromEntries(state || [])), errors, location_key: locationKey, }) @@ -122,11 +124,12 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { options: UpdateProcessedEntityOptions, ): Promise { const tx = txOpaque as Knex.Transaction; - const { id, errors } = options; + const { id, errors, resultHash } = options; await tx('refresh_state') .update({ errors, + result_hash: resultHash, }) .where('entity_id', id); } @@ -492,6 +495,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { processedEntity: i.processed_entity ? (JSON.parse(i.processed_entity) as Entity) : undefined, + 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 395694131d..d735607554 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; + 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 78199900e4..c93b90084d 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; + resultHash: string; state?: Map; errors?: string; relations: EntityRelationSpec[]; @@ -44,6 +45,7 @@ export type UpdateProcessedEntityOptions = { export type UpdateProcessedEntityErrorsOptions = { id: string; errors?: string; + resultHash: string; }; export type RefreshStateItem = { @@ -51,6 +53,7 @@ export type RefreshStateItem = { entityRef: string; unprocessedEntity: Entity; processedEntity?: Entity; + resultHash: string; nextUpdateAt: string; lastDiscoveryAt: string; // remove? state: Map;