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

Merge pull request #10544 from backstage/freben/stitch-more

Browse Source 
stitch sources of refs that are removed too
This commit is contained in:
Fredrik Adelöw
2022-03-31 15:37:50 +02:00
committed by GitHub
parent dcc3aae979 934bd6b8cb
commit 83ff7f68b6
6 changed files with 220 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/breezy-schools-compete.md
+7
View File
@@ -0,0 +1,7 @@
---
'@backstage/plugin-catalog-backend': patch
---
Fixed an issue where sometimes entities would have stale relations "stuck" and
not getting removed as expected, after the other end of the relation had stopped
referring to them.
plugins/catalog-backend/src/database/DefaultProcessingDatabase.test.ts
+32 -2
View File
@@ -221,7 +221,7 @@ describe('Default Processing Database', () => {
},
];
await db.transaction(tx =>
let updateResult = await db.transaction(tx =>
db.updateProcessedEntity(tx, {
id,
processedEntity,
@@ -231,7 +231,7 @@ describe('Default Processing Database', () => {
}),
);
const savedRelations = await knex<DbRelationsRow>('relations')
let savedRelations = await knex<DbRelationsRow>('relations')
.where({ originating_entity_id: id })
.select();
expect(savedRelations.length).toBe(1);
@@ -241,6 +241,36 @@ describe('Default Processing Database', () => {
type: 'memberOf',
target_entity_ref: 'component:default/foo',
});
expect(updateResult.previous.relations).toEqual([]);
updateResult = await db.transaction(tx =>
db.updateProcessedEntity(tx, {
id,
processedEntity,
resultHash: '',
relations: relations,
deferredEntities: [],
}),
);
savedRelations = await knex<DbRelationsRow>('relations')
.where({ originating_entity_id: id })
.select();
expect(savedRelations.length).toBe(1);
expect(savedRelations[0]).toEqual({
originating_entity_id: id,
source_entity_ref: 'component:default/foo',
type: 'memberOf',
target_entity_ref: 'component:default/foo',
});
expect(updateResult.previous.relations).toEqual([
{
originating_entity_id: expect.any(String),
source_entity_ref: 'component:default/foo',
type: 'memberOf',
target_entity_ref: 'component:default/foo',
},
]);
},
60_000,
);
plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts
+21 -4
View File
@@ -68,7 +68,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
async updateProcessedEntity(
txOpaque: Transaction,
options: UpdateProcessedEntityOptions,
): Promise<void> {
): Promise<{ previous: { relations: DbRelationsRow[] } }> {
const tx = txOpaque as Knex.Transaction;
const {
id,
@@ -108,9 +108,20 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
});
// Delete old relations
await tx<DbRelationsRow>('relations')
.where({ originating_entity_id: id })
.delete();
let previousRelationRows: DbRelationsRow[];
if (tx.client.config.client.includes('sqlite3')) {
previousRelationRows = await tx<DbRelationsRow>('relations')
.select('*')
.where({ originating_entity_id: id });
await tx<DbRelationsRow>('relations')
.where({ originating_entity_id: id })
.delete();
} else {
previousRelationRows = await tx<DbRelationsRow>('relations')
.where({ originating_entity_id: id })
.delete()
.returning('*');
}
// Batch insert new relations
const relationRows: DbRelationsRow[] = relations.map(
@@ -126,6 +137,12 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
this.deduplicateRelations(relationRows),
BATCH_SIZE,
);
return {
previous: {
relations: previousRelationRows,
},
};
}
async updateProcessedEntityErrors(
plugins/catalog-backend/src/database/types.ts
+2 -1
View File
@@ -19,6 +19,7 @@ import { JsonObject } from '@backstage/types';
import { DateTime } from 'luxon';
import { EntityRelationSpec } from '../api';
import { DeferredEntity } from '../processing/types';
import { DbRelationsRow } from './tables';
/**
* An abstraction for transactions of the underlying database technology.
@@ -125,7 +126,7 @@ export interface ProcessingDatabase {
updateProcessedEntity(
txOpaque: Transaction,
options: UpdateProcessedEntityOptions,
): Promise<void>;
): Promise<{ previous: { relations: DbRelationsRow[] } }>;
/**
* Updates the cache associated with an entity.
plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts
+127
View File
@@ -91,6 +91,10 @@ describe('DefaultCatalogProcessingEngine', () => {
},
],
});
db.listParents.mockResolvedValue({ entityRefs: [] });
db.updateProcessedEntity.mockResolvedValue({
previous: { relations: [] },
});
await engine.start();
await waitForExpect(() => {
@@ -153,6 +157,10 @@ describe('DefaultCatalogProcessingEngine', () => {
},
],
});
db.listParents.mockResolvedValue({ entityRefs: [] });
db.updateProcessedEntity.mockImplementation(async () => ({
previous: { relations: [] },
}));
await engine.start();
await waitForExpect(() => {
@@ -213,6 +221,10 @@ describe('DefaultCatalogProcessingEngine', () => {
items: [{ ...refreshState, resultHash: 'NOT RIGHT' }],
})
.mockResolvedValue({ items: [] });
db.updateProcessedEntity.mockImplementation(async () => ({
previous: { relations: [] },
}));
await engine.start();
await waitForExpect(() => {
@@ -285,6 +297,10 @@ describe('DefaultCatalogProcessingEngine', () => {
items: [refreshState],
})
.mockResolvedValue({ items: [] });
db.listParents.mockResolvedValue({ entityRefs: [] });
db.updateProcessedEntity.mockImplementation(async () => ({
previous: { relations: [] },
}));
await waitForExpect(() => {
expect(db.updateEntityCache).toBeCalledTimes(1);
@@ -319,4 +335,115 @@ describe('DefaultCatalogProcessingEngine', () => {
await engine.stop();
});
it('should stitch both the previous and new sources when relations change', async () => {
const engine = new DefaultCatalogProcessingEngine(
getVoidLogger(),
db,
orchestrator,
stitcher,
() => hash,
100,
);
db.transaction.mockImplementation(cb => cb((() => {}) as any));
const entity = {
apiVersion: '1',
kind: 'k',
metadata: { name: 'me', namespace: 'ns' },
};
const processableEntity = {
entityRef: 'foo',
id: '1',
unprocessedEntity: entity,
resultHash: '',
state: [] as any,
nextUpdateAt: DateTime.now(),
lastDiscoveryAt: DateTime.now(),
};
db.listParents.mockResolvedValue({ entityRefs: [] });
db.getProcessableEntities
.mockResolvedValueOnce({
items: [processableEntity],
})
.mockResolvedValueOnce({
items: [processableEntity],
});
db.updateProcessedEntity
.mockImplementationOnce(async () => ({
previous: { relations: [] },
}))
.mockImplementationOnce(async () => ({
previous: {
relations: [
{
originating_entity_id: '',
type: 't',
source_entity_ref: 'k:ns/other1',
target_entity_ref: 'k:ns/me',
},
{
originating_entity_id: '',
type: 't',
source_entity_ref: 'k:ns/other2',
target_entity_ref: 'k:ns/me',
},
],
},
}));
orchestrator.process
.mockResolvedValueOnce({
ok: true,
completedEntity: entity,
relations: [
{
type: 't',
source: { kind: 'k', namespace: 'ns', name: 'other1' },
target: { kind: 'k', namespace: 'ns', name: 'me' },
},
{
type: 't',
source: { kind: 'k', namespace: 'ns', name: 'other2' },
target: { kind: 'k', namespace: 'ns', name: 'me' },
},
],
errors: [],
deferredEntities: [],
state: {},
})
.mockResolvedValueOnce({
ok: true,
completedEntity: entity,
relations: [
{
type: 't',
source: { kind: 'k', namespace: 'ns', name: 'other2' },
target: { kind: 'k', namespace: 'ns', name: 'me' },
},
{
type: 't',
source: { kind: 'k', namespace: 'ns', name: 'other3' },
target: { kind: 'k', namespace: 'ns', name: 'me' },
},
],
errors: [],
deferredEntities: [],
state: {},
});
await engine.start();
await waitForExpect(() => {
expect(stitcher.stitch).toBeCalledTimes(2);
});
expect([...stitcher.stitch.mock.calls[0][0]]).toEqual(
expect.arrayContaining(['k:ns/me', 'k:ns/other1', 'k:ns/other2']),
);
expect([...stitcher.stitch.mock.calls[1][0]]).toEqual(
expect.arrayContaining(['k:ns/me', 'k:ns/other1', 'k:ns/other3']),
);
await engine.stop();
});
});
plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts
+31 -12
View File
@@ -169,24 +169,43 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
}
result.completedEntity.metadata.uid = id;
let oldRelationSources: Set<string>;
await this.processingDatabase.transaction(async tx => {
await this.processingDatabase.updateProcessedEntity(tx, {
id,
processedEntity: result.completedEntity,
resultHash,
errors: errorsString,
relations: result.relations,
deferredEntities: result.deferredEntities,
locationKey,
});
const { previous } =
await this.processingDatabase.updateProcessedEntity(tx, {
id,
processedEntity: result.completedEntity,
resultHash,
errors: errorsString,
relations: result.relations,
deferredEntities: result.deferredEntities,
locationKey,
});
oldRelationSources = new Set(
previous.relations.map(r => r.source_entity_ref),
);
});
const newRelationSources = new Set<string>(
result.relations.map(relation =>
stringifyEntityRef(relation.source),
),
);
const setOfThingsToStitch = new Set<string>([
stringifyEntityRef(result.completedEntity),
...result.relations.map(relation =>
stringifyEntityRef(relation.source),
),
]);
newRelationSources.forEach(r => {
if (!oldRelationSources.has(r)) {
setOfThingsToStitch.add(r);
}
});
oldRelationSources!.forEach(r => {
if (!newRelationSources.has(r)) {
setOfThingsToStitch.add(r);
}
});
await this.stitcher.stitch(setOfThingsToStitch);
track.markSuccessfulWithChanges(setOfThingsToStitch.size);