paulononato/backstage
1
0
Fork 0
Code Issues Pull Requests Actions 39 Packages Projects Releases Wiki Activity

catalog-backend: fix catalog processing not replacing all existing references

Browse Source 
Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>
This commit is contained in:
Patrik Oldsberg
2025-01-16 12:02:41 +01:00
parent e57fd0c63d
commit 11d0b7526a
3 changed files with 208 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/fair-mangos-sleep.md
+1 -1
View File
@@ -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
plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts
+5 -2
View File
@@ -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<DbRefreshStateReferencesRow>('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',
plugins/catalog-backend/src/tests/integration.test.ts
+202 -23
View File
@@ -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<string>) {
@@ -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<DbRefreshStateReferencesRow>('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<DbRefreshStateReferencesRow>('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),
});
});
});