From 1db090c8d118463eb6de954a09480df8b009c5d4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Mon, 27 Sep 2021 09:40:40 +0200 Subject: [PATCH 1/4] rename toAdd to toUpsert to clarify what it actually does MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- .../database/DefaultProcessingDatabase.ts | 22 +++++++++---------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index 3dfb4beb80..b20c890cbd 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -156,7 +156,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { ): Promise { const tx = txOpaque as Knex.Transaction; - const { toAdd, toRemove } = await this.createDelta(tx, options); + const { toUpsert, toRemove } = await this.createDelta(tx, options); if (toRemove.length) { // TODO(freben): Batch split, to not hit variable limits? @@ -273,14 +273,14 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { ); } - if (toAdd.length) { - for (const { entity, locationKey } of toAdd) { + if (toUpsert.length) { + for (const { entity, locationKey } of toUpsert) { const entityRef = stringifyEntityRef(entity); try { - let ok = await this.insertUnprocessedEntity(tx, entity, locationKey); + let ok = await this.updateUnprocessedEntity(tx, entity, locationKey); if (!ok) { - ok = await this.updateUnprocessedEntity(tx, entity, locationKey); + ok = await this.insertUnprocessedEntity(tx, entity, locationKey); } if (ok) { @@ -558,10 +558,10 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { private async createDelta( tx: Knex.Transaction, options: ReplaceUnprocessedEntitiesOptions, - ): Promise<{ toAdd: DeferredEntity[]; toRemove: string[] }> { + ): Promise<{ toUpsert: DeferredEntity[]; toRemove: string[] }> { if (options.type === 'delta') { return { - toAdd: options.added, + toUpsert: options.added, toRemove: options.removed.map(e => stringifyEntityRef(e.entity)), }; } @@ -586,7 +586,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { ); const newRefsSet = new Set(items.map(item => item.ref)); - const toAdd = new Array(); + const toUpsert = new Array(); const toRemove = oldRefs .map(row => row.target_entity_ref) .filter(ref => !newRefsSet.has(ref)); @@ -594,15 +594,15 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { for (const item of items) { if (!oldRefsSet.has(item.ref)) { // Add any entity that does not exist in the database - toAdd.push(item.deferred); + toUpsert.push(item.deferred); } else if (oldRefsSet.get(item.ref) !== item.deferred.locationKey) { // Remove and then re-add any entity that exists, but with a different location key toRemove.push(item.ref); - toAdd.push(item.deferred); + toUpsert.push(item.deferred); } } - return { toAdd, toRemove }; + return { toUpsert, toRemove }; } /** From d0209d8e11a5f271166d9b06fe8c70b0b7a21016 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Mon, 27 Sep 2021 13:01:46 +0200 Subject: [PATCH 2/4] Store and maintain a hash of unprocessed items as well MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- ...0925102509_add_refresh_state_input_hash.js | 38 ++++++++++ .../DefaultProcessingDatabase.test.ts | 2 + .../database/DefaultProcessingDatabase.ts | 69 +++++++++++++++---- .../src/next/database/tables.ts | 1 + .../catalog-backend/src/next/database/util.ts | 25 +++++++ 5 files changed, 122 insertions(+), 13 deletions(-) create mode 100644 plugins/catalog-backend/migrations/20210925102509_add_refresh_state_input_hash.js create mode 100644 plugins/catalog-backend/src/next/database/util.ts 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'); +} From 3d10360c8276308498edacaba4c55b996c9e0fcd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Mon, 27 Sep 2021 17:00:31 +0200 Subject: [PATCH 3/4] add changeset MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- .changeset/breezy-foxes-wonder.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changeset/breezy-foxes-wonder.md diff --git a/.changeset/breezy-foxes-wonder.md b/.changeset/breezy-foxes-wonder.md new file mode 100644 index 0000000000..57b5c52892 --- /dev/null +++ b/.changeset/breezy-foxes-wonder.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend': patch +--- + +When issuing a `full` update from an entity provider, entities with updates are now properly persisted. From bf650f86031a4338c578d8acaca5674122ec7431 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Tue, 28 Sep 2021 13:31:49 +0200 Subject: [PATCH 4/4] add a test too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- .../DefaultProcessingDatabase.test.ts | 94 +++++++++++++++++++ 1 file changed, 94 insertions(+) diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts index 90367bbf01..6148325581 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts @@ -998,6 +998,100 @@ describe('Default Processing Database', () => { }, 60_000, ); + + it.each(databases.eachSupportedId())( + 'should support replacing modified entities during a full update, %p', + async databaseId => { + const { knex, db } = await createDatabase(databaseId); + + await db.transaction(async tx => { + await db.replaceUnprocessedEntities(tx, { + type: 'full', + sourceKey: 'lols', + items: [ + { + entity: { + apiVersion: '1', + kind: 'Component', + metadata: { name: 'a' }, + spec: { marker: 'WILL_CHANGE' }, + } as Entity, + locationKey: 'file:///tmp/a', + }, + { + entity: { + apiVersion: '1', + kind: 'Component', + metadata: { name: 'b' }, + spec: { marker: 'NEVER_CHANGES' }, + } as Entity, + locationKey: 'file:///tmp/b', + }, + ], + }); + }); + + let state = await knex('refresh_state').select(); + expect(state).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + entity_ref: 'component:default/a', + location_key: 'file:///tmp/a', + unprocessed_entity: expect.stringContaining('WILL_CHANGE'), + }), + expect.objectContaining({ + entity_ref: 'component:default/b', + location_key: 'file:///tmp/b', + unprocessed_entity: expect.stringContaining('NEVER_CHANGES'), + }), + ]), + ); + + await db.transaction(async tx => { + await db.replaceUnprocessedEntities(tx, { + type: 'full', + sourceKey: 'lols', + items: [ + { + entity: { + apiVersion: '1', + kind: 'Component', + metadata: { name: 'a' }, + spec: { marker: 'HAS_CHANGED' }, + } as Entity, + locationKey: 'file:///tmp/a', + }, + { + entity: { + apiVersion: '1', + kind: 'Component', + metadata: { name: 'b' }, + spec: { marker: 'NEVER_CHANGES' }, + } as Entity, + locationKey: 'file:///tmp/b', + }, + ], + }); + }); + + state = await knex('refresh_state').select(); + expect(state).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + entity_ref: 'component:default/a', + location_key: 'file:///tmp/a', + unprocessed_entity: expect.stringContaining('HAS_CHANGED'), + }), + expect.objectContaining({ + entity_ref: 'component:default/b', + location_key: 'file:///tmp/b', + unprocessed_entity: expect.stringContaining('NEVER_CHANGES'), + }), + ]), + ); + }, + 60_000, + ); }); describe('getProcessableEntities', () => {