diff --git a/.changeset/itchy-rats-pump.md b/.changeset/itchy-rats-pump.md new file mode 100644 index 0000000000..3b9073da77 --- /dev/null +++ b/.changeset/itchy-rats-pump.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend': patch +--- + +Compute deltas more efficiently, which generally leads to less wasted processing cycles diff --git a/plugins/catalog-backend/src/database/DefaultProviderDatabase.test.ts b/plugins/catalog-backend/src/database/DefaultProviderDatabase.test.ts index 0218d25a18..adab1531b0 100644 --- a/plugins/catalog-backend/src/database/DefaultProviderDatabase.test.ts +++ b/plugins/catalog-backend/src/database/DefaultProviderDatabase.test.ts @@ -19,13 +19,14 @@ import { TestDatabaseId, TestDatabases, } from '@backstage/backend-test-utils'; -import { Entity } from '@backstage/catalog-model'; +import { Entity, stringifyEntityRef } from '@backstage/catalog-model'; import { Knex } from 'knex'; import * as uuid from 'uuid'; import { DefaultProviderDatabase } from './DefaultProviderDatabase'; import { applyDatabaseMigrations } from './migrations'; import { DbRefreshStateReferencesRow, DbRefreshStateRow } from './tables'; import { LoggerService } from '@backstage/backend-plugin-api'; +import { generateStableHash } from './util'; jest.setTimeout(60_000); @@ -690,13 +691,8 @@ describe('DefaultProviderDatabase', () => { it.each(databases.eachSupportedId())( 'should successfully fall back from batch to individual mode on conflicts, %p', async databaseId => { - const fakeLogger = { - debug: jest.fn(), - }; - const { knex, db } = await createDatabase( - databaseId, - fakeLogger as any, - ); + const fakeLogger = mockServices.logger.mock(); + const { knex, db } = await createDatabase(databaseId, fakeLogger); await createLocations(knex, ['component:default/a']); @@ -744,11 +740,8 @@ describe('DefaultProviderDatabase', () => { it.each(databases.eachSupportedId())( 'should gracefully handle accidental duplicate refresh state references when deletion happens during a full sync, %p', async databaseId => { - const fakeLogger = { debug: jest.fn() }; - const { knex, db } = await createDatabase( - databaseId, - fakeLogger as any, - ); + const fakeLogger = mockServices.logger.mock(); + const { knex, db } = await createDatabase(databaseId, fakeLogger); await createLocations(knex, ['component:default/a']); @@ -773,5 +766,162 @@ describe('DefaultProviderDatabase', () => { expect(state).toEqual([]); }, ); + + it.each(databases.eachSupportedId())( + 'should properly translate deltas into add/update/remove, %p', + async databaseId => { + const fakeLogger = mockServices.logger.mock(); + const { knex, db } = await createDatabase(databaseId, fakeLogger); + + const entity1Before: Entity = { + apiVersion: '1', + kind: 'k', + metadata: { namespace: 'ns', name: 'n1' }, + }; + const entity1After: Entity = { + ...entity1Before, + apiVersion: '2', + }; + + const entity2Before: Entity = { + apiVersion: '1', + kind: 'k', + metadata: { namespace: 'ns', name: 'n2' }, + }; + const entity2After: Entity = { + ...entity2Before, + apiVersion: '2', + }; + + const entity3: Entity = { + apiVersion: '1', + kind: 'k', + metadata: { namespace: 'ns', name: 'n3' }, + }; + + const entity4: Entity = { + apiVersion: '1', + kind: 'k', + metadata: { namespace: 'ns', name: 'n4' }, + }; + + await insertRefreshStateRow(knex, { + entity_id: 'id1', + entity_ref: stringifyEntityRef(entity1Before), + last_discovery_at: new Date(), + next_update_at: new Date(), + errors: '[]', + unprocessed_entity: JSON.stringify(entity1Before), + unprocessed_hash: generateStableHash(entity1Before), + }); + await insertRefRow(knex, { + source_key: 'my-provider', + target_entity_ref: stringifyEntityRef(entity1Before), + }); + await insertRefreshStateRow(knex, { + entity_id: 'id2', + entity_ref: stringifyEntityRef(entity2Before), + last_discovery_at: new Date(), + next_update_at: new Date(), + errors: '[]', + unprocessed_entity: JSON.stringify(entity2Before), + unprocessed_hash: generateStableHash(entity2After), // lie about the hash! + }); + await insertRefRow(knex, { + source_key: 'my-provider', + target_entity_ref: stringifyEntityRef(entity2Before), + }); + await insertRefreshStateRow(knex, { + entity_id: 'id4', + entity_ref: stringifyEntityRef(entity4), + last_discovery_at: new Date(), + next_update_at: new Date(), + errors: '[]', + unprocessed_entity: JSON.stringify(entity4), + unprocessed_hash: generateStableHash(entity4), + }); + await insertRefRow(knex, { + source_key: 'my-provider', + target_entity_ref: stringifyEntityRef(entity4), + }); + + await db.transaction(async tx => { + await db.replaceUnprocessedEntities(tx, { + type: 'delta', + sourceKey: 'my-provider', + added: [ + // we used the right hashes for entity1, so this will turn into an update + { entity: entity1After }, + // we lied about the hash for entity2, so this will become a no-op + { entity: entity2After }, + // this didn't exist, so will become an add + { entity: entity3 }, + ], + removed: [{ entityRef: stringifyEntityRef(entity4) }], + }); + }); + + const state = await knex('refresh_state') + .select(['entity_ref', 'unprocessed_entity', 'unprocessed_hash']) + .orderBy('entity_ref'); + + expect(state).toEqual([ + { + entity_ref: stringifyEntityRef(entity1After), + unprocessed_entity: JSON.stringify(entity1After), + unprocessed_hash: generateStableHash(entity1After), + }, + { + entity_ref: stringifyEntityRef(entity2After), + unprocessed_entity: JSON.stringify(entity2Before), // didn't change + unprocessed_hash: generateStableHash(entity2After), + }, + { + entity_ref: stringifyEntityRef(entity3), + unprocessed_entity: JSON.stringify(entity3), + unprocessed_hash: generateStableHash(entity3), + }, + ]); + }, + ); + + it.each(databases.eachSupportedId())( + 'can handle large deltas without exploding, %p', + async databaseId => { + const fakeLogger = mockServices.logger.mock(); + const { knex, db } = await createDatabase(databaseId, fakeLogger); + + const count = 10000; + const padded = (n: number) => String(n).padStart(8, '0'); + + const entities = Array.from({ length: count }, (_, i) => ({ + entity: { + apiVersion: '1', + kind: 'k', + metadata: { namespace: 'ns', name: padded(i) }, + }, + })); + + await db.transaction(async tx => { + await db.replaceUnprocessedEntities(tx, { + type: 'delta', + sourceKey: 'my-provider', + added: entities, + removed: [], + }); + }); + + const state = await knex('refresh_state') + .select(['entity_ref', 'unprocessed_entity', 'unprocessed_hash']) + .orderBy('entity_ref'); + + expect(state).toHaveLength(count); + expect(state[0]).toEqual({ + entity_ref: stringifyEntityRef(entities[0].entity), + unprocessed_entity: JSON.stringify(entities[0].entity), + unprocessed_hash: generateStableHash(entities[0].entity), + }); + }, + ); }); }); diff --git a/plugins/catalog-backend/src/database/DefaultProviderDatabase.ts b/plugins/catalog-backend/src/database/DefaultProviderDatabase.ts index 94026fbfa0..bbb24560be 100644 --- a/plugins/catalog-backend/src/database/DefaultProviderDatabase.ts +++ b/plugins/catalog-backend/src/database/DefaultProviderDatabase.ts @@ -75,10 +75,10 @@ export class DefaultProviderDatabase implements ProviderDatabase { } async replaceUnprocessedEntities( - txOpaque: Transaction, + txOpaque: Knex | Transaction, options: ReplaceUnprocessedEntitiesOptions, ): Promise { - const tx = txOpaque as Knex.Transaction; + const tx = txOpaque as Knex | Knex.Transaction; const { toAdd, toUpsert, toRemove } = await this.createDelta(tx, options); if (toRemove.length) { @@ -205,7 +205,7 @@ export class DefaultProviderDatabase implements ProviderDatabase { } private async createDelta( - tx: Knex.Transaction, + tx: Knex | Knex.Transaction, options: ReplaceUnprocessedEntitiesOptions, ): Promise<{ toAdd: { deferred: DeferredEntity; hash: string }[]; @@ -213,14 +213,32 @@ export class DefaultProviderDatabase implements ProviderDatabase { toRemove: string[]; }> { if (options.type === 'delta') { - return { - toAdd: [], - toUpsert: options.added.map(e => ({ - deferred: e, - hash: generateStableHash(e.entity), - })), - toRemove: options.removed.map(e => e.entityRef), - }; + const toAdd = new Array<{ deferred: DeferredEntity; hash: string }>(); + const toUpsert = new Array<{ deferred: DeferredEntity; hash: string }>(); + const toRemove = options.removed.map(e => e.entityRef); + + for (const chunk of lodash.chunk(options.added, 1000)) { + const entityRefs = chunk.map(e => stringifyEntityRef(e.entity)); + const rows = await tx('refresh_state') + .select(['entity_ref', 'unprocessed_hash']) + .whereIn('entity_ref', entityRefs); + const oldHashes = new Map( + rows.map(row => [row.entity_ref, row.unprocessed_hash]), + ); + + chunk.forEach((deferred, i) => { + const entityRef = entityRefs[i]; + const newHash = generateStableHash(deferred.entity); + const oldHash = oldHashes.get(entityRef); + if (oldHash === undefined) { + toAdd.push({ deferred, hash: newHash }); + } else if (newHash !== oldHash) { + toUpsert.push({ deferred, hash: newHash }); + } + }); + } + + return { toAdd, toUpsert, toRemove }; } // Grab all of the existing references from the same source, and their locationKeys as well diff --git a/plugins/catalog-backend/src/database/operations/refreshState/checkLocationKeyConflict.ts b/plugins/catalog-backend/src/database/operations/refreshState/checkLocationKeyConflict.ts index 97d376f9a4..cf6c920d1d 100644 --- a/plugins/catalog-backend/src/database/operations/refreshState/checkLocationKeyConflict.ts +++ b/plugins/catalog-backend/src/database/operations/refreshState/checkLocationKeyConflict.ts @@ -24,7 +24,7 @@ import { DbRefreshStateRow } from '../../tables'; * @returns The conflicting key if there is one. */ export async function checkLocationKeyConflict(options: { - tx: Knex.Transaction; + tx: Knex | Knex.Transaction; entityRef: string; locationKey?: string; }): Promise { diff --git a/plugins/catalog-backend/src/database/operations/refreshState/insertUnprocessedEntity.ts b/plugins/catalog-backend/src/database/operations/refreshState/insertUnprocessedEntity.ts index 195dbe1f2c..54067c3f2e 100644 --- a/plugins/catalog-backend/src/database/operations/refreshState/insertUnprocessedEntity.ts +++ b/plugins/catalog-backend/src/database/operations/refreshState/insertUnprocessedEntity.ts @@ -28,7 +28,7 @@ import { * true if successful and false if there was a conflict. */ export async function insertUnprocessedEntity(options: { - tx: Knex.Transaction; + tx: Knex | Knex.Transaction; entity: Entity; hash: string; locationKey?: string; diff --git a/plugins/catalog-backend/src/database/operations/refreshState/updateUnprocessedEntity.ts b/plugins/catalog-backend/src/database/operations/refreshState/updateUnprocessedEntity.ts index e26db4429b..ced6b0d54d 100644 --- a/plugins/catalog-backend/src/database/operations/refreshState/updateUnprocessedEntity.ts +++ b/plugins/catalog-backend/src/database/operations/refreshState/updateUnprocessedEntity.ts @@ -25,7 +25,7 @@ import { DbRefreshStateRow } from '../../tables'; * Updating the entity will also cause it to be scheduled for immediate processing. */ export async function updateUnprocessedEntity(options: { - tx: Knex.Transaction; + tx: Knex | Knex.Transaction; entity: Entity; hash: string; locationKey?: string; diff --git a/plugins/catalog-backend/src/tests/performance/providerDeltaMutations.test.ts b/plugins/catalog-backend/src/tests/performance/providerDeltaMutations.test.ts new file mode 100644 index 0000000000..6e95c2535a --- /dev/null +++ b/plugins/catalog-backend/src/tests/performance/providerDeltaMutations.test.ts @@ -0,0 +1,115 @@ +/* + * Copyright 2024 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 { TestDatabases, mockServices } from '@backstage/backend-test-utils'; +import { Knex } from 'knex'; +import { applyDatabaseMigrations } from '../../database/migrations'; +import { describePerformanceTest, performanceTraceEnabled } from './lib/env'; +import { DefaultProviderDatabase } from '../../database/DefaultProviderDatabase'; +import { DeferredEntity } from '@backstage/plugin-catalog-node'; + +// #region Helpers + +jest.setTimeout(600_000); + +const databases = TestDatabases.create({ + ids: [/* 'MYSQL_8', */ 'POSTGRES_16', /* 'POSTGRES_12',*/ 'SQLITE_3'], + disableDocker: false, +}); + +const traceLog: typeof console.log = performanceTraceEnabled + ? console.log + : () => {}; + +// #endregion +// #region Tests + +describePerformanceTest('providerDeltaMutations', () => { + let knex: Knex; + + describe.each(databases.eachSupportedId())('%p', databaseId => { + beforeAll(async () => { + knex = await databases.init(databaseId); + await applyDatabaseMigrations(knex); + }); + + afterAll(async () => { + await knex.destroy(); + }); + + it.each([200, 50_000])( + 'inserts and then overwrites identical sets using delta mutations, batch size %p', + async batchSize => { + const sut = new DefaultProviderDatabase({ + database: knex, + logger: mockServices.logger.mock(), + }); + + const sourceKey = 'source-key'; + const entities: DeferredEntity[] = Array.from( + { length: batchSize }, + (_, i) => ({ + entity: { + apiVersion: 'backstage.io/v1alpha1', + kind: 'Component', + metadata: { + name: `component-${i}`, + namespace: 'default', + }, + spec: { + type: 'service', + lifecycle: 'production', + }, + }, + }), + ); + + let start = Date.now(); + await sut.replaceUnprocessedEntities(knex, { + type: 'delta', + sourceKey, + added: entities, + removed: [], + }); + let perSecond = Math.round(batchSize / ((Date.now() - start) / 1000)); + traceLog( + `${databaseId} inserted ${perSecond} entities per second at a batch size of ${batchSize}`, + ); + + start = Date.now(); + const rounds = 20; + for (let i = 0; i < rounds; i++) { + await sut.replaceUnprocessedEntities(knex, { + type: 'delta', + sourceKey, + added: entities, + removed: [], + }); + } + perSecond = Math.round( + (batchSize * rounds) / ((Date.now() - start) / 1000), + ); + traceLog( + `${databaseId} updated ${perSecond} entities per second at a batch size of ${batchSize}`, + ); + + expect(true).toBe(true); + }, + ); + }); +}); + +// #endregion