From 11d0b7526afc737a977fc3a882e751e37350ddd9 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Thu, 16 Jan 2025 12:02:41 +0100 Subject: [PATCH] catalog-backend: fix catalog processing not replacing all existing references Signed-off-by: Patrik Oldsberg --- .changeset/fair-mangos-sleep.md | 2 +- .../src/database/DefaultProcessingDatabase.ts | 7 +- .../src/tests/integration.test.ts | 225 ++++++++++++++++-- 3 files changed, 208 insertions(+), 26 deletions(-) diff --git a/.changeset/fair-mangos-sleep.md b/.changeset/fair-mangos-sleep.md index e2b24d441d..8f0a6371c3 100644 --- a/.changeset/fair-mangos-sleep.md +++ b/.changeset/fair-mangos-sleep.md @@ -2,4 +2,4 @@ '@backstage/plugin-catalog-backend': patch --- -Cleanup `refresh_state_references` for providers that are no longer in control of a `refresh_state` row for entity +Cleanup `refresh_state_references` for entity processors and providers that are no longer in control of a `refresh_state` row for entity diff --git a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts index ee836ff038..bb5c38de18 100644 --- a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts @@ -383,9 +383,12 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { } } - // Replace all references for the originating entity or source and then create new ones + // Lastly, replace refresh state references for the originating entity and any successfully added entities await tx('refresh_state_references') - .andWhere({ source_entity_ref: options.sourceEntityRef }) + // Remove all existing references from the originating entity + .where({ source_entity_ref: options.sourceEntityRef }) + // And remove any existing references to entities that we're inserting new references for + .orWhereIn('target_entity_ref', stateReferences) .delete(); await tx.batchInsert( 'refresh_state_references', diff --git a/plugins/catalog-backend/src/tests/integration.test.ts b/plugins/catalog-backend/src/tests/integration.test.ts index bc4a68e5c3..0c8e475b39 100644 --- a/plugins/catalog-backend/src/tests/integration.test.ts +++ b/plugins/catalog-backend/src/tests/integration.test.ts @@ -31,7 +31,8 @@ import { } from '@backstage/plugin-catalog-node'; import { PermissionEvaluator } from '@backstage/plugin-permission-common'; import { createHash } from 'crypto'; -import knexFactory, { Knex } from 'knex'; +import { Knex } from 'knex'; +import merge from 'lodash/merge'; import { EntitiesCatalog } from '../catalog/types'; import { DefaultCatalogDatabase } from '../database/DefaultCatalogDatabase'; import { DefaultProcessingDatabase } from '../database/DefaultProcessingDatabase'; @@ -55,7 +56,7 @@ import { mockServices } from '@backstage/backend-test-utils'; import { LoggerService } from '@backstage/backend-plugin-api'; import { DatabaseManager } from '@backstage/backend-common'; import { entitiesResponseToObjects } from '../service/response'; -import { DbRefreshStateReferencesRow } from '../database/tables'; +import { deleteOrphanedEntities } from '../database/operations/util/deleteOrphanedEntities'; const voidLogger = mockServices.logger.mock(); @@ -197,6 +198,7 @@ class TestHarness { readonly #refresh: RefreshService; readonly #provider: TestProvider; readonly #proxyProgressTracker: ProxyProgressTracker; + readonly #db: Knex; static async create(options?: { disableRelationsCompatibility?: boolean; @@ -326,6 +328,7 @@ class TestHarness { refresh, provider, proxyProgressTracker, + db, ); } @@ -335,12 +338,14 @@ class TestHarness { refresh: RefreshService, provider: TestProvider, proxyProgressTracker: ProxyProgressTracker, + db: Knex, ) { this.#catalog = catalog; this.#engine = engine; this.#refresh = refresh; this.#provider = provider; this.#proxyProgressTracker = proxyProgressTracker; + this.#db = db; } async process(entityRefs?: Set) { @@ -382,6 +387,55 @@ class TestHarness { async refresh(options: RefreshOptions) { return this.#refresh.refresh(options); } + + async removeOrphanedEntities() { + await deleteOrphanedEntities({ + knex: this.#db, + strategy: { mode: 'immediate' }, + }); + } + + async getRefreshState(): Promise< + Record< + string, + { + id: string; + unprocessedEntity: Entity; + processedEntity: Entity; + locationKey: string | null; + } + > + > { + const result = await this.#db('refresh_state').select('*'); + return Object.fromEntries( + result.map(r => [ + r.entity_ref, + { + id: r.entity_id, + unprocessedEntity: JSON.parse(r.unprocessed_entity), + processedEntity: r.processed_entity + ? JSON.parse(r.processed_entity) + : undefined, + locationKey: r.location_key, + }, + ]), + ); + } + + async getRefreshStateReferences(): Promise< + Array<{ + sourceKey: string | null; + sourceEntityRef: string | null; + targetEntityRef: string; + }> + > { + const result = await this.#db('refresh_state_references').select('*'); + return result.map(r => ({ + sourceKey: r.source_key ?? undefined, + sourceEntityRef: r.source_entity_ref ?? undefined, + targetEntityRef: r.target_entity_ref, + })); + } } describe('Catalog Backend Integration', () => { @@ -847,12 +901,6 @@ describe('Catalog Backend Integration', () => { }); it('should replace any refresh_state_references that are dangling after claiming an entityRef with locationKey', async () => { - const db = knexFactory({ - client: 'better-sqlite3', - connection: ':memory:', - useNullAsDefault: true, - }); - const firstProvider = new TestProvider(); firstProvider.getProviderName = () => 'first'; @@ -861,7 +909,6 @@ describe('Catalog Backend Integration', () => { const harness = await TestHarness.create({ additionalProviders: [firstProvider, secondProvider], - db, }); await firstProvider.getConnection().applyMutation({ @@ -933,15 +980,11 @@ describe('Catalog Backend Integration', () => { }), }); - await expect( - db('refresh_state_references') - .where({ target_entity_ref: 'component:default/component-1' }) - .select(), - ).resolves.toEqual([ - expect.objectContaining({ - source_key: 'second', - target_entity_ref: 'component:default/component-1', - }), + await expect(harness.getRefreshStateReferences()).resolves.toEqual([ + { + sourceKey: 'second', + targetEntityRef: 'component:default/component-1', + }, ]); await secondProvider.getConnection().applyMutation({ @@ -970,11 +1013,12 @@ describe('Catalog Backend Integration', () => { await expect(harness.process()).resolves.toEqual({}); - await expect( - db('refresh_state_references') - .where({ target_entity_ref: 'component:default/component-1' }) - .select(), - ).resolves.toEqual([]); + await expect(harness.getRefreshStateReferences()).resolves.toEqual([ + { + sourceKey: 'second', + targetEntityRef: 'component:default/component-2', + }, + ]); await expect(harness.getOutputEntities()).resolves.toEqual({ 'component:default/component-2': expect.objectContaining({ @@ -985,4 +1029,139 @@ describe('Catalog Backend Integration', () => { }), }); }); + + function withOutputFields(entity: Entity) { + return { + ...entity, + metadata: { + ...entity.metadata, + etag: expect.any(String), + uid: expect.any(String), + }, + relations: [], + }; + } + + it('should fully replace existing entities when emitting override entities during processing', async () => { + const baseEntity = { + apiVersion: 'backstage.io/v1alpha1', + kind: 'Component', + metadata: { + annotations: { + 'backstage.io/managed-by-location': 'url:.', + 'backstage.io/managed-by-origin-location': 'url:.', + }, + }, + }; + const entityA = merge({ metadata: { name: 'a' } }, baseEntity); + const entityB = merge({ metadata: { name: 'b' } }, baseEntity); + const entityBOverride = merge({ metadata: { override: true } }, entityB); + + const processEntity = jest.fn( + async ( + entity: Entity, + _location: LocationSpec, + _emit: CatalogProcessorEmit, + ) => entity, + ); + const harness = await TestHarness.create({ processEntity }); + + processEntity.mockImplementation(async (entity, location, emit) => { + if (entity.metadata.name === entityA.metadata.name) { + emit(processingResult.entity(location, entityBOverride)); + } + return entity; + }); + + // A and B are added to the catalog, but the processor emits B' from A that overrides B + await harness.setInputEntities([entityA, entityB]); + await expect(harness.process()).resolves.toEqual({}); + + // Expect to find A and B' in the catalog + await expect(harness.getRefreshStateReferences()).resolves.toEqual([ + { + sourceKey: 'test', + targetEntityRef: 'component:default/a', + }, + { + sourceEntityRef: 'component:default/a', + targetEntityRef: 'component:default/b', + }, + ]); + await expect(harness.getRefreshState()).resolves.toEqual({ + 'component:default/a': expect.objectContaining({ + locationKey: null, + unprocessedEntity: entityA, + }), + 'component:default/b': expect.objectContaining({ + locationKey: 'url:.', + unprocessedEntity: entityBOverride, + }), + }); + await expect(harness.getOutputEntities()).resolves.toEqual({ + 'component:default/a': withOutputFields(entityA), + 'component:default/b': withOutputFields(entityBOverride), + }); + + // Stop emitting B' from A, then do a full sync with A and B + processEntity.mockImplementation(async entity => entity); + await harness.setInputEntities([entityA, entityB]); + + // At this point we should still have A and B' in the catalog + await expect(harness.getRefreshStateReferences()).resolves.toEqual([ + { + sourceKey: 'test', + targetEntityRef: 'component:default/a', + }, + { + sourceEntityRef: 'component:default/a', + targetEntityRef: 'component:default/b', + }, + ]); + await expect(harness.getRefreshState()).resolves.toEqual({ + 'component:default/a': expect.objectContaining({ + locationKey: null, + unprocessedEntity: entityA, + }), + 'component:default/b': expect.objectContaining({ + locationKey: 'url:.', + unprocessedEntity: entityBOverride, + }), + }); + await expect(harness.getOutputEntities()).resolves.toEqual({ + 'component:default/a': withOutputFields(entityA), + 'component:default/b': withOutputFields(entityBOverride), + }); + + // Once we process, B' should be orphaned + await expect(harness.process()).resolves.toEqual({}); + // This is expected to remove B' + await harness.removeOrphanedEntities(); + + // At this point only A is left in the catalog + await expect(harness.getRefreshStateReferences()).resolves.toEqual([ + { + sourceKey: 'test', + targetEntityRef: 'component:default/a', + }, + ]); + await expect(harness.getRefreshState()).resolves.toEqual({ + 'component:default/a': expect.objectContaining({ + locationKey: null, + unprocessedEntity: entityA, + }), + }); + await expect(harness.getOutputEntities()).resolves.toEqual({ + 'component:default/a': withOutputFields(entityA), + }); + + // Next time the provider runs and does a full sync we should now be able to add back B + await harness.setInputEntities([entityA, entityB]); + await expect(harness.process()).resolves.toEqual({}); + + await expect(harness.getOutputEntities()).resolves.toEqual({ + 'component:default/a': withOutputFields(entityA), + 'component:default/b': withOutputFields(entityB), + }); + }); });