diff --git a/plugins/catalog-backend/migrations/20210925102509_add_refresh_state_input_hash.js b/plugins/catalog-backend/migrations/20210925102509_add_refresh_state_input_hash.js new file mode 100644 index 0000000000..7d7a4ac578 --- /dev/null +++ b/plugins/catalog-backend/migrations/20210925102509_add_refresh_state_input_hash.js @@ -0,0 +1,38 @@ +/* + * 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('unprocessed_hash') + .nullable() + .comment('A hash of the unprocessed contents, used to detect changes'); + }); +}; + +/** + * @param {import('knex').Knex} knex + */ +exports.down = async function down(knex) { + await knex.schema.alterTable('refresh_state', table => { + table.dropColumn('unprocessed_hash'); + }); +}; diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts index 2a28951e6b..90367bbf01 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts @@ -30,6 +30,7 @@ import { } from './tables'; import { createRandomRefreshInterval } from '../refresh'; import { timestampToDateTime } from './conversion'; +import { generateStableHash } from './util'; describe('Default Processing Database', () => { const defaultLogger = getVoidLogger(); @@ -309,6 +310,7 @@ describe('Default Processing Database', () => { entity_id: id, entity_ref: 'location:default/fakelocation', unprocessed_entity: '{}', + unprocessed_hash: generateStableHash({} as any), processed_entity: '{}', errors: '[]', next_update_at: '2021-04-01 13:37:00', diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index b20c890cbd..7b32bec5e9 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -41,6 +41,7 @@ import { ListAncestorsResult, UpdateEntityCacheOptions, } from './types'; +import { generateStableHash } from './util'; // The number of items that are sent per batch to the database layer, when // doing .batchInsert calls to knex. This needs to be low enough to not cause @@ -274,13 +275,26 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { } if (toUpsert.length) { - for (const { entity, locationKey } of toUpsert) { + for (const { + deferred: { entity, locationKey }, + hash, + } of toUpsert) { const entityRef = stringifyEntityRef(entity); try { - let ok = await this.updateUnprocessedEntity(tx, entity, locationKey); + let ok = await this.updateUnprocessedEntity( + tx, + entity, + hash, + locationKey, + ); if (!ok) { - ok = await this.insertUnprocessedEntity(tx, entity, locationKey); + ok = await this.insertUnprocessedEntity( + tx, + entity, + hash, + locationKey, + ); } if (ok) { @@ -448,6 +462,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { private async updateUnprocessedEntity( tx: Knex.Transaction, entity: Entity, + hash: string, locationKey?: string, ): Promise { const entityRef = stringifyEntityRef(entity); @@ -456,6 +471,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { const refreshResult = await tx('refresh_state') .update({ unprocessed_entity: serializedEntity, + unprocessed_hash: hash, location_key: locationKey, last_discovery_at: tx.fn.now(), // We only get to this point if a processed entity actually had any changes, or @@ -483,6 +499,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { private async insertUnprocessedEntity( tx: Knex.Transaction, entity: Entity, + hash: string, locationKey?: string, ): Promise { const entityRef = stringifyEntityRef(entity); @@ -493,6 +510,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { entity_id: uuid(), entity_ref: entityRef, unprocessed_entity: serializedEntity, + unprocessed_hash: hash, errors: '', location_key: locationKey, next_update_at: tx.fn.now(), @@ -558,10 +576,16 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { private async createDelta( tx: Knex.Transaction, options: ReplaceUnprocessedEntitiesOptions, - ): Promise<{ toUpsert: DeferredEntity[]; toRemove: string[] }> { + ): Promise<{ + toUpsert: { deferred: DeferredEntity; hash: string }[]; + toRemove: string[]; + }> { if (options.type === 'delta') { return { - toUpsert: options.added, + toUpsert: options.added.map(e => ({ + deferred: e, + hash: generateStableHash(e.entity), + })), toRemove: options.removed.map(e => stringifyEntityRef(e.entity)), }; } @@ -570,35 +594,51 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { const oldRefs = await tx( 'refresh_state_references', ) - .where({ source_key: options.sourceKey }) .leftJoin('refresh_state', { target_entity_ref: 'entity_ref', }) - .select(['target_entity_ref', 'location_key']); + .where({ source_key: options.sourceKey }) + .select({ + target_entity_ref: 'refresh_state_references.target_entity_ref', + location_key: 'refresh_state.location_key', + unprocessed_hash: 'refresh_state.unprocessed_hash', + }); const items = options.items.map(deferred => ({ deferred, ref: stringifyEntityRef(deferred.entity), + hash: generateStableHash(deferred.entity), })); const oldRefsSet = new Map( - oldRefs.map(r => [r.target_entity_ref, r.location_key]), + oldRefs.map(r => [ + r.target_entity_ref, + { + locationKey: r.location_key, + oldEntityHash: r.unprocessed_hash, + }, + ]), ); const newRefsSet = new Set(items.map(item => item.ref)); - const toUpsert = new Array(); + const toUpsert = new Array<{ deferred: DeferredEntity; hash: string }>(); const toRemove = oldRefs .map(row => row.target_entity_ref) .filter(ref => !newRefsSet.has(ref)); for (const item of items) { - if (!oldRefsSet.has(item.ref)) { + const oldRef = oldRefsSet.get(item.ref); + const upsertItem = { deferred: item.deferred, hash: item.hash }; + if (!oldRef) { // Add any entity that does not exist in the database - toUpsert.push(item.deferred); - } else if (oldRefsSet.get(item.ref) !== item.deferred.locationKey) { + toUpsert.push(upsertItem); + } else if (oldRef.locationKey !== item.deferred.locationKey) { // Remove and then re-add any entity that exists, but with a different location key toRemove.push(item.ref); - toUpsert.push(item.deferred); + toUpsert.push(upsertItem); + } else if (oldRef.oldEntityHash !== item.hash) { + // Entities with modifications should be pushed through too + toUpsert.push(upsertItem); } } @@ -626,10 +666,12 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { // their entity ref. for (const { entity, locationKey } of options.entities) { const entityRef = stringifyEntityRef(entity); + const hash = generateStableHash(entity); const updated = await this.updateUnprocessedEntity( tx, entity, + hash, locationKey, ); if (updated) { @@ -640,6 +682,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { const inserted = await this.insertUnprocessedEntity( tx, entity, + hash, locationKey, ); if (inserted) { diff --git a/plugins/catalog-backend/src/next/database/tables.ts b/plugins/catalog-backend/src/next/database/tables.ts index b40964d225..e064b17d7b 100644 --- a/plugins/catalog-backend/src/next/database/tables.ts +++ b/plugins/catalog-backend/src/next/database/tables.ts @@ -24,6 +24,7 @@ export type DbRefreshStateRow = { entity_id: string; entity_ref: string; unprocessed_entity: string; + unprocessed_hash?: string; processed_entity?: string; result_hash?: string; cache?: string; diff --git a/plugins/catalog-backend/src/next/database/util.ts b/plugins/catalog-backend/src/next/database/util.ts new file mode 100644 index 0000000000..771fab7fba --- /dev/null +++ b/plugins/catalog-backend/src/next/database/util.ts @@ -0,0 +1,25 @@ +/* + * 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. + */ + +import { Entity } from '@backstage/catalog-model'; +import { createHash } from 'crypto'; +import stableStringify from 'fast-json-stable-stringify'; + +export function generateStableHash(entity: Entity) { + return createHash('sha1') + .update(stableStringify({ ...entity })) + .digest('hex'); +}