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

fix(catalog-backend): remove immediate mode stitching

Browse Source 
Remove the immediate stitching strategy from the catalog-backend plugin.
All stitching now uses the deferred mode exclusively, which processes
entities asynchronously via a worker queue. This simplifies the stitching
subsystem by removing dead-weight code paths and the strategy parameter
threading, and fixes a bug where deleteWithEagerPruningOfChildren
hardcoded immediate-mode semantics regardless of configured mode.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Signed-off-by: Fredrik Adelöw <freben@gmail.com>
This commit is contained in:
Fredrik Adelöw
2026-05-11 23:20:55 +02:00
parent 74c5fbbafd
commit dc7678cdcc
21 changed files with 772 additions and 1365 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/remove-immediate-stitching.md
+5
View File
@@ -0,0 +1,5 @@
---
'@backstage/plugin-catalog-backend': patch
---
Removed the immediate mode stitching strategy. All stitching now uses the deferred mode, which processes entities asynchronously via a worker queue. If your configuration includes `catalog.stitchingStrategy.mode: 'immediate'`, it will be ignored with a deprecation warning. The `pollingInterval` and `stitchTimeout` settings continue to work as before.
plugins/catalog-backend/config.d.ts
Vendored
+9 -18
View File
@@ -175,25 +175,16 @@ export interface Config {
orphanProviderStrategy?: 'keep' | 'delete';
/**
* The strategy to use when stitching together the final entities. The default mode is "deferred".
* The strategy to use when stitching together the final entities.
*/
stitchingStrategy?:
| {
/**
* Perform stitching in-band immediately when needed.
*
* @deprecated Immediate mode stitching has been deprecated and will be removed in a future release. Migrate to deferred mode (the default).
*/
mode: 'immediate';
}
| {
/** Defer stitching to be performed asynchronously */
mode: 'deferred';
/** Polling interval for tasks in seconds */
pollingInterval?: HumanDuration | string;
/** How long to wait for a stitch to complete before giving up in seconds */
stitchTimeout?: HumanDuration | string;
};
stitchingStrategy?: {
/** @deprecated Immediate mode has been removed. This field is ignored. */
mode?: string;
/** Polling interval for tasks in seconds */
pollingInterval?: HumanDuration | string;
/** How long to wait for a stitch to complete before giving up in seconds */
stitchTimeout?: HumanDuration | string;
};
/**
* The strategy to use when there is a conflict with a location being registered.
plugins/catalog-backend/src/database/operations/provider/deleteWithEagerPruningOfChildren.test.ts
+2 -3
View File
@@ -85,10 +85,9 @@ describe.each(databases.eachSupportedId())(
}
async function entitiesMarkedForStitching(knex: Knex) {
const rows = await knex<DbRefreshStateRow>('refresh_state')
const rows = await knex('stitch_queue')
.orderBy('entity_ref')
.select('entity_ref')
.where('result_hash', '=', 'force-stitching');
.select('entity_ref');
return rows.map(r => r.entity_ref);
}
plugins/catalog-backend/src/database/operations/provider/deleteWithEagerPruningOfChildren.ts
+10 -20
View File
@@ -16,11 +16,8 @@
import { Knex } from 'knex';
import lodash from 'lodash';
import {
DbFinalEntitiesRow,
DbRefreshStateReferencesRow,
DbRefreshStateRow,
} from '../../tables';
import { DbRefreshStateReferencesRow } from '../../tables';
import { markForStitching } from '../stitcher/markForStitching';
/**
* Given a number of entity refs originally created by a given entity provider
@@ -57,6 +54,10 @@ export async function deleteWithEagerPruningOfChildren(options: {
.delete()
.from('refresh_state')
.whereIn('entity_ref', refsToDelete);
await knex
.delete()
.from('stitch_queue')
.whereIn('entity_ref', refsToDelete);
}
// Delete the references that originate only from this entity provider. Note
@@ -249,19 +250,8 @@ async function markEntitiesAffectedByDeletionForStitching(options: {
.whereIn('relations.target_entity_ref', entityRefs)
.then(rows => rows.map(row => row.entity_id));
for (const ids of lodash.chunk(affectedIds, 1000)) {
await knex
.table<DbFinalEntitiesRow>('final_entities')
.update({
hash: 'force-stitching',
})
.whereIn('entity_id', ids);
await knex
.table<DbRefreshStateRow>('refresh_state')
.update({
result_hash: 'force-stitching',
next_update_at: knex.fn.now(),
})
.whereIn('entity_id', ids);
}
await markForStitching({
knex,
entityIds: affectedIds,
});
}
plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts
+262 -534
View File
@@ -17,550 +17,278 @@
import { TestDatabases } from '@backstage/backend-test-utils';
import { applyDatabaseMigrations } from '../../migrations';
import { markForStitching } from './markForStitching';
import {
DbFinalEntitiesRow,
DbRefreshStateRow,
DbStitchQueueRow,
} from '../../tables';
import { DbRefreshStateRow, DbStitchQueueRow } from '../../tables';
jest.setTimeout(60_000);
const databases = TestDatabases.create();
describe.each(databases.eachSupportedId())(
'markForStitching, %p',
databaseId => {
it('marks the right rows in deferred mode', async () => {
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
it.each(databases.eachSupportedId())(
'marks the right rows %p',
async databaseId => {
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
await knex<DbRefreshStateRow>('refresh_state').insert([
{
entity_id: '1',
entity_ref: 'k:ns/one',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
{
entity_id: '2',
entity_ref: 'k:ns/two',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
{
entity_id: '3',
entity_ref: 'k:ns/three',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
{
entity_id: '4',
entity_ref: 'k:ns/four',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
]);
// Entity 4 has an existing stitch_queue row with old stitch data
await knex<DbStitchQueueRow>('stitch_queue').insert([
{
entity_ref: 'k:ns/four',
stitch_ticket: 'old',
next_stitch_at: '1971-01-01T00:00:00.000',
},
]);
await knex<DbRefreshStateRow>('refresh_state').insert([
{
entity_id: '1',
entity_ref: 'k:ns/one',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
{
entity_id: '2',
entity_ref: 'k:ns/two',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
{
entity_id: '3',
entity_ref: 'k:ns/three',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
{
entity_id: '4',
entity_ref: 'k:ns/four',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
]);
// Entity 4 has an existing stitch_queue row with old stitch data
await knex<DbStitchQueueRow>('stitch_queue').insert([
{
entity_ref: 'k:ns/four',
stitch_ticket: 'old',
next_stitch_at: '1971-01-01T00:00:00.000',
},
]);
async function result() {
return knex<DbStitchQueueRow>('stitch_queue')
.select('entity_ref', 'next_stitch_at', 'stitch_ticket')
.orderBy('entity_ref', 'asc');
}
// Initially only entity 4 has a stitch_queue row
const original = await result();
expect(original).toEqual([
{
entity_ref: 'k:ns/four',
next_stitch_at: expect.anything(),
stitch_ticket: 'old',
},
]);
// Calling with empty set should not create any new rows
await markForStitching({
knex,
strategy: {
mode: 'deferred',
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
},
entityRefs: new Set(),
});
await expect(result()).resolves.toEqual([
{
entity_ref: 'k:ns/four',
next_stitch_at: expect.anything(),
stitch_ticket: 'old',
},
]);
// Mark entity 1 - should create a new stitch_queue row
await markForStitching({
knex,
strategy: {
mode: 'deferred',
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
},
entityRefs: new Set(['k:ns/one']),
});
await expect(result()).resolves.toEqual([
{
entity_ref: 'k:ns/four',
next_stitch_at: expect.anything(),
stitch_ticket: 'old',
},
{
entity_ref: 'k:ns/one',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
]);
// Mark entity 2 - should create another new stitch_queue row
await markForStitching({
knex,
strategy: {
mode: 'deferred',
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
},
entityRefs: ['k:ns/two'],
});
await expect(result()).resolves.toEqual([
{
entity_ref: 'k:ns/four',
next_stitch_at: expect.anything(),
stitch_ticket: 'old',
},
{
entity_ref: 'k:ns/one',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
{
entity_ref: 'k:ns/two',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
]);
// Mark entities 3 and 4 by ID - entity 3 creates new row, entity 4 updates existing
await markForStitching({
knex,
strategy: {
mode: 'deferred',
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
},
entityIds: ['3', '4'],
});
await expect(result()).resolves.toEqual([
{
entity_ref: 'k:ns/four',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
{
entity_ref: 'k:ns/one',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
{
entity_ref: 'k:ns/three',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
{
entity_ref: 'k:ns/two',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
]);
// Entity 4's ticket should have been updated (was 'old', now something else)
const final = await result();
const entity4Final = final.find(r => r.entity_ref === 'k:ns/four');
expect(entity4Final?.stitch_ticket).not.toEqual('old');
});
it('marks the right rows in immediate mode', async () => {
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
await knex<DbRefreshStateRow>('refresh_state').insert([
{
entity_id: '1',
entity_ref: 'k:ns/one',
unprocessed_entity: '{}',
processed_entity: '{}',
result_hash: 'old',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
{
entity_id: '2',
entity_ref: 'k:ns/two',
unprocessed_entity: '{}',
processed_entity: '{}',
result_hash: 'old',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
{
entity_id: '3',
entity_ref: 'k:ns/three',
unprocessed_entity: '{}',
processed_entity: '{}',
result_hash: 'old',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
{
entity_id: '4',
entity_ref: 'k:ns/four',
unprocessed_entity: '{}',
processed_entity: '{}',
result_hash: 'old',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
]);
await knex<DbFinalEntitiesRow>('final_entities').insert([
{
entity_id: '1',
final_entity: '{}',
entity_ref: 'k:ns/one',
hash: 'old',
},
{
entity_id: '2',
final_entity: '{}',
entity_ref: 'k:ns/two',
hash: 'old',
},
{
entity_id: '3',
final_entity: '{}',
entity_ref: 'k:ns/three',
hash: 'old',
},
{
entity_id: '4',
final_entity: '{}',
entity_ref: 'k:ns/four',
hash: 'old',
},
]);
async function result() {
return knex<DbRefreshStateRow>('refresh_state')
.leftJoin(
'final_entities',
'final_entities.entity_id',
'refresh_state.entity_id',
)
.select({
entity_id: 'refresh_state.entity_id',
next_update_at: 'refresh_state.next_update_at',
refresh_state_hash: 'refresh_state.result_hash',
final_entities_hash: 'final_entities.hash',
})
.orderBy('entity_id', 'asc');
}
// Ensure that now() isn't evaluating to the same thing
await new Promise(resolve => setTimeout(resolve, 1100));
const original = await result();
await markForStitching({
knex,
strategy: { mode: 'immediate' },
entityRefs: new Set(),
});
await expect(result()).resolves.toEqual([
{
entity_id: '1',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
{
entity_id: '2',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
{
entity_id: '3',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
{
entity_id: '4',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
]);
await markForStitching({
knex,
strategy: { mode: 'immediate' },
entityRefs: new Set(['k:ns/one']),
});
await expect(result()).resolves.toEqual([
{
entity_id: '1',
next_update_at: expect.anything(),
refresh_state_hash: 'force-stitching',
final_entities_hash: 'force-stitching',
},
{
entity_id: '2',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
{
entity_id: '3',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
{
entity_id: '4',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
]);
await markForStitching({
knex,
strategy: { mode: 'immediate' },
entityRefs: ['k:ns/two'],
});
await expect(result()).resolves.toEqual([
{
entity_id: '1',
next_update_at: expect.anything(),
refresh_state_hash: 'force-stitching',
final_entities_hash: 'force-stitching',
},
{
entity_id: '2',
next_update_at: expect.anything(),
refresh_state_hash: 'force-stitching',
final_entities_hash: 'force-stitching',
},
{
entity_id: '3',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
{
entity_id: '4',
next_update_at: expect.anything(),
refresh_state_hash: 'old',
final_entities_hash: 'old',
},
]);
await markForStitching({
knex,
strategy: { mode: 'immediate' },
entityIds: ['3', '4'],
});
await expect(result()).resolves.toEqual([
{
entity_id: '1',
next_update_at: expect.anything(),
refresh_state_hash: 'force-stitching',
final_entities_hash: 'force-stitching',
},
{
entity_id: '2',
next_update_at: expect.anything(),
refresh_state_hash: 'force-stitching',
final_entities_hash: 'force-stitching',
},
{
entity_id: '3',
next_update_at: expect.anything(),
refresh_state_hash: 'force-stitching',
final_entities_hash: 'force-stitching',
},
{
entity_id: '4',
next_update_at: expect.anything(),
refresh_state_hash: 'force-stitching',
final_entities_hash: 'force-stitching',
},
]);
// It overwrites timers
const final = await result();
for (let i = 0; i < final.length; ++i) {
expect(original[i].next_update_at).not.toEqual(final[i].next_update_at);
}
});
it('reproduces deadlock scenario when concurrent transactions update overlapping entity sets', async () => {
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
// Setup test data with multiple entities
const entityRefs = [
'k:ns/entity-a',
'k:ns/entity-b',
'k:ns/entity-c',
'k:ns/entity-d',
'k:ns/entity-e',
'k:ns/entity-f',
];
await knex<DbRefreshStateRow>('refresh_state').insert(
entityRefs.map((ref, i) => ({
entity_id: `${i + 1}`,
entity_ref: ref,
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
})),
);
// This test attempts to reproduce the deadlock by running concurrent transactions
// that update overlapping sets of entities in different orders
const errorResults = [];
for (let attempt = 0; attempt < 10; attempt++) {
// Transaction 1: Update entities A, B, C, D, E
const transaction1 = knex.transaction(async trx => {
await markForStitching({
knex: trx,
strategy: {
mode: 'deferred',
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
},
entityRefs: [
'k:ns/entity-a',
'k:ns/entity-b',
'k:ns/entity-c',
'k:ns/entity-d',
'k:ns/entity-e',
],
});
// Add a small delay to increase chance of collision
await new Promise(resolve => setTimeout(resolve, 10));
await markForStitching({
knex: trx,
strategy: {
mode: 'deferred',
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
},
entityRefs: ['k:ns/entity-f'],
});
});
// Transaction 2: Update entities F, E, D, C, B (reverse order)
const transaction2 = knex.transaction(async trx => {
await markForStitching({
knex: trx,
strategy: {
mode: 'deferred',
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
},
entityRefs: [
'k:ns/entity-f',
'k:ns/entity-e',
'k:ns/entity-d',
'k:ns/entity-c',
'k:ns/entity-b',
],
});
// Add a small delay to increase chance of collision
await new Promise(resolve => setTimeout(resolve, 10));
await markForStitching({
knex: trx,
strategy: {
mode: 'deferred',
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
},
entityRefs: ['k:ns/entity-a'],
});
});
// Run both transactions concurrently to create potential deadlock
errorResults.push(
Promise.allSettled([transaction1, transaction2]).then(results =>
results
.filter(r => r.status === 'rejected')
.map(r => (r as PromiseRejectedResult).reason),
),
);
}
const allResults = await Promise.all(errorResults);
const deadlockErrors = allResults
.flat()
.filter(
error =>
error?.code === '40P01' ||
error?.message?.includes('deadlock detected') ||
error?.message?.includes('deadlock'),
);
expect(deadlockErrors).toEqual([]);
// Verify final state - all entities should have been marked for stitching
const finalState = await knex<DbStitchQueueRow>('stitch_queue')
async function result() {
return knex<DbStitchQueueRow>('stitch_queue')
.select('entity_ref', 'next_stitch_at', 'stitch_ticket')
.orderBy('entity_ref');
.orderBy('entity_ref', 'asc');
}
expect(finalState.length).toBeGreaterThan(0);
finalState.forEach(row => {
expect(row.next_stitch_at).not.toBeNull();
expect(row.stitch_ticket).not.toBeNull();
// Initially only entity 4 has a stitch_queue row
const original = await result();
expect(original).toEqual([
{
entity_ref: 'k:ns/four',
next_stitch_at: expect.anything(),
stitch_ticket: 'old',
},
]);
// Calling with empty set should not create any new rows
await markForStitching({
knex,
entityRefs: new Set(),
});
await expect(result()).resolves.toEqual([
{
entity_ref: 'k:ns/four',
next_stitch_at: expect.anything(),
stitch_ticket: 'old',
},
]);
// Mark entity 1 - should create a new stitch_queue row
await markForStitching({
knex,
entityRefs: new Set(['k:ns/one']),
});
await expect(result()).resolves.toEqual([
{
entity_ref: 'k:ns/four',
next_stitch_at: expect.anything(),
stitch_ticket: 'old',
},
{
entity_ref: 'k:ns/one',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
]);
// Mark entity 2 - should create another new stitch_queue row
await markForStitching({
knex,
entityRefs: ['k:ns/two'],
});
await expect(result()).resolves.toEqual([
{
entity_ref: 'k:ns/four',
next_stitch_at: expect.anything(),
stitch_ticket: 'old',
},
{
entity_ref: 'k:ns/one',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
{
entity_ref: 'k:ns/two',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
]);
// Mark entities 3 and 4 by ID - entity 3 creates new row, entity 4 updates existing
await markForStitching({
knex,
entityIds: ['3', '4'],
});
await expect(result()).resolves.toEqual([
{
entity_ref: 'k:ns/four',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
{
entity_ref: 'k:ns/one',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
{
entity_ref: 'k:ns/three',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
{
entity_ref: 'k:ns/two',
next_stitch_at: expect.anything(),
stitch_ticket: expect.anything(),
},
]);
// Entity 4's ticket should have been updated (was 'old', now something else)
const final = await result();
const entity4Final = final.find(r => r.entity_ref === 'k:ns/four');
expect(entity4Final?.stitch_ticket).not.toEqual('old');
},
);
it.each(databases.eachSupportedId())(
'reproduces deadlock scenario when concurrent transactions update overlapping entity sets %p',
async databaseId => {
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
// Setup test data with multiple entities
const entityRefs = [
'k:ns/entity-a',
'k:ns/entity-b',
'k:ns/entity-c',
'k:ns/entity-d',
'k:ns/entity-e',
'k:ns/entity-f',
];
await knex<DbRefreshStateRow>('refresh_state').insert(
entityRefs.map((ref, i) => ({
entity_id: `${i + 1}`,
entity_ref: ref,
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
})),
);
// This test attempts to reproduce the deadlock by running concurrent transactions
// that update overlapping sets of entities in different orders
const errorResults = [];
for (let attempt = 0; attempt < 10; attempt++) {
// Transaction 1: Update entities A, B, C, D, E
const transaction1 = knex.transaction(async trx => {
await markForStitching({
knex: trx,
entityRefs: [
'k:ns/entity-a',
'k:ns/entity-b',
'k:ns/entity-c',
'k:ns/entity-d',
'k:ns/entity-e',
],
});
// Add a small delay to increase chance of collision
await new Promise(resolve => setTimeout(resolve, 10));
await markForStitching({
knex: trx,
entityRefs: ['k:ns/entity-f'],
});
});
// Transaction 2: Update entities F, E, D, C, B (reverse order)
const transaction2 = knex.transaction(async trx => {
await markForStitching({
knex: trx,
entityRefs: [
'k:ns/entity-f',
'k:ns/entity-e',
'k:ns/entity-d',
'k:ns/entity-c',
'k:ns/entity-b',
],
});
// Add a small delay to increase chance of collision
await new Promise(resolve => setTimeout(resolve, 10));
await markForStitching({
knex: trx,
entityRefs: ['k:ns/entity-a'],
});
});
// Run both transactions concurrently to create potential deadlock
errorResults.push(
Promise.allSettled([transaction1, transaction2]).then(results =>
results
.filter(r => r.status === 'rejected')
.map(r => (r as PromiseRejectedResult).reason),
),
);
}
const allResults = await Promise.all(errorResults);
const deadlockErrors = allResults
.flat()
.filter(
error =>
error?.code === '40P01' ||
error?.message?.includes('deadlock detected') ||
error?.message?.includes('deadlock'),
);
expect(deadlockErrors).toEqual([]);
// Verify final state - all entities should have been marked for stitching
const finalState = await knex<DbStitchQueueRow>('stitch_queue')
.select('entity_ref', 'next_stitch_at', 'stitch_ticket')
.orderBy('entity_ref');
expect(finalState.length).toBeGreaterThan(0);
finalState.forEach(row => {
expect(row.next_stitch_at).not.toBeNull();
expect(row.stitch_ticket).not.toBeNull();
});
},
);
plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts
+39 -86
View File
@@ -17,12 +17,7 @@
import { Knex } from 'knex';
import splitToChunks from 'lodash/chunk';
import { randomUUID as uuid } from 'node:crypto';
import { StitchingStrategy } from '../../../stitching/types';
import {
DbFinalEntitiesRow,
DbRefreshStateRow,
DbStitchQueueRow,
} from '../../tables';
import { DbRefreshStateRow, DbStitchQueueRow } from '../../tables';
import { retryOnDeadlock } from '../../util';
const UPDATE_CHUNK_SIZE = 100; // Smaller chunks reduce contention
@@ -35,96 +30,54 @@ const UPDATE_CHUNK_SIZE = 100; // Smaller chunks reduce contention
*/
export async function markForStitching(options: {
knex: Knex | Knex.Transaction;
strategy: StitchingStrategy;
entityRefs?: Iterable<string>;
entityIds?: Iterable<string>;
}): Promise<void> {
const entityRefs = sortSplit(options.entityRefs);
const entityIds = sortSplit(options.entityIds);
const knex = options.knex;
const mode = options.strategy.mode;
if (mode === 'immediate') {
for (const chunk of entityRefs) {
await knex
.table<DbFinalEntitiesRow>('final_entities')
.update({
hash: 'force-stitching',
})
.whereIn('entity_ref', chunk);
await retryOnDeadlock(async () => {
await knex
.table<DbRefreshStateRow>('refresh_state')
.update({
result_hash: 'force-stitching',
next_update_at: knex.fn.now(),
})
.whereIn('entity_ref', chunk);
}, knex);
}
// It's OK that this is shared across stitch_queue rows; it just needs to
// be uniquely generated for every new stitch request.
const ticket = uuid();
for (const chunk of entityIds) {
await knex
.table<DbFinalEntitiesRow>('final_entities')
.update({
hash: 'force-stitching',
})
for (const chunk of entityRefs) {
await retryOnDeadlock(async () => {
if (chunk.length > 0) {
await knex<DbStitchQueueRow>('stitch_queue')
.insert(
chunk.map(ref => ({
entity_ref: ref,
stitch_ticket: ticket,
next_stitch_at: knex.fn.now(),
})),
)
.onConflict('entity_ref')
.merge(['next_stitch_at', 'stitch_ticket']);
}
}, knex);
}
for (const chunk of entityIds) {
await retryOnDeadlock(async () => {
// Look up entity_refs from refresh_state by entity_id
const refreshStateRows = await knex<DbRefreshStateRow>('refresh_state')
.select('entity_ref')
.whereIn('entity_id', chunk);
await retryOnDeadlock(async () => {
await knex
.table<DbRefreshStateRow>('refresh_state')
.update({
result_hash: 'force-stitching',
next_update_at: knex.fn.now(),
})
.whereIn('entity_id', chunk);
}, knex);
}
} else if (mode === 'deferred') {
// It's OK that this is shared across stitch_queue rows; it just needs to
// be uniquely generated for every new stitch request.
const ticket = uuid();
for (const chunk of entityRefs) {
await retryOnDeadlock(async () => {
if (chunk.length > 0) {
await knex<DbStitchQueueRow>('stitch_queue')
.insert(
chunk.map(ref => ({
entity_ref: ref,
stitch_ticket: ticket,
next_stitch_at: knex.fn.now(),
})),
)
.onConflict('entity_ref')
.merge(['next_stitch_at', 'stitch_ticket']);
}
}, knex);
}
for (const chunk of entityIds) {
await retryOnDeadlock(async () => {
// Look up entity_refs from refresh_state by entity_id
const refreshStateRows = await knex<DbRefreshStateRow>('refresh_state')
.select('entity_ref')
.whereIn('entity_id', chunk);
if (refreshStateRows.length > 0) {
await knex<DbStitchQueueRow>('stitch_queue')
.insert(
refreshStateRows.map(row => ({
entity_ref: row.entity_ref,
stitch_ticket: ticket,
next_stitch_at: knex.fn.now(),
})),
)
.onConflict('entity_ref')
.merge(['next_stitch_at', 'stitch_ticket']);
}
}, knex);
}
} else {
throw new Error(`Unknown stitching strategy mode ${mode}`);
if (refreshStateRows.length > 0) {
await knex<DbStitchQueueRow>('stitch_queue')
.insert(
refreshStateRows.map(row => ({
entity_ref: row.entity_ref,
stitch_ticket: ticket,
next_stitch_at: knex.fn.now(),
})),
)
.onConflict('entity_ref')
.merge(['next_stitch_at', 'stitch_ticket']);
}
}, knex);
}
}
plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts
+268 -279
View File
@@ -31,301 +31,292 @@ jest.setTimeout(60_000);
const databases = TestDatabases.create();
describe.each(databases.eachSupportedId())(
'performStitching, %p',
databaseId => {
it.each(databases.eachSupportedId())(
'runs the happy path for %p',
async databaseId => {
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
const logger = mockServices.logger.mock();
// NOTE(freben): Testing the deferred path since it's a superset of the immediate one
it('runs the happy path in deferred mode', async () => {
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
let entities: DbFinalEntitiesRow[];
let entity: Entity;
let entities: DbFinalEntitiesRow[];
let entity: Entity;
await knex<DbRefreshStateRow>('refresh_state').insert([
{
entity_id: 'my-id',
entity_ref: 'k:ns/n',
unprocessed_entity: JSON.stringify({}),
processed_entity: JSON.stringify({
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
},
spec: {
k: 'v',
},
}),
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
]);
await knex<DbRefreshStateReferencesRow>(
'refresh_state_references',
).insert([{ source_key: 'a', target_entity_ref: 'k:ns/n' }]);
await knex<DbRelationsRow>('relations').insert([
{
originating_entity_id: 'my-id',
source_entity_ref: 'k:ns/n',
type: 'looksAt',
target_entity_ref: 'k:ns/other',
},
// handles and ignores duplicates
{
originating_entity_id: 'my-id',
source_entity_ref: 'k:ns/n',
type: 'looksAt',
target_entity_ref: 'k:ns/other',
},
]);
await knex<DbRefreshStateRow>('refresh_state').insert([
await markForStitching({
knex,
entityRefs: ['k:ns/n'],
});
await performStitching({
knex,
logger,
entityRef: 'k:ns/n',
stitchTicket: (
await knex('stitch_queue')
.where('entity_ref', 'k:ns/n')
.select('stitch_ticket')
.first()
)?.stitch_ticket,
});
entities = await knex<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
entity = JSON.parse(entities[0].final_entity!);
expect(entity).toEqual({
relations: [
{
type: 'looksAt',
targetRef: 'k:ns/other',
},
],
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
etag: expect.any(String),
uid: 'my-id',
},
spec: {
k: 'v',
},
});
expect(entity.metadata.etag).toEqual(entities[0].hash);
const last_updated_at = entities[0].last_updated_at;
expect(last_updated_at).not.toBeNull();
const firstHash = entities[0].hash;
const search = await knex<DbSearchRow>('search');
expect(search).toEqual(
expect.arrayContaining([
{
entity_id: 'my-id',
entity_ref: 'k:ns/n',
unprocessed_entity: JSON.stringify({}),
processed_entity: JSON.stringify({
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
},
spec: {
k: 'v',
},
}),
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
key: 'relations.looksat',
original_value: 'k:ns/other',
value: 'k:ns/other',
},
]);
await knex<DbRefreshStateReferencesRow>(
'refresh_state_references',
).insert([{ source_key: 'a', target_entity_ref: 'k:ns/n' }]);
await knex<DbRelationsRow>('relations').insert([
{
originating_entity_id: 'my-id',
source_entity_ref: 'k:ns/n',
type: 'looksAt',
target_entity_ref: 'k:ns/other',
entity_id: 'my-id',
key: 'apiversion',
original_value: 'a',
value: 'a',
},
// handles and ignores duplicates
{
originating_entity_id: 'my-id',
source_entity_ref: 'k:ns/n',
type: 'looksAt',
target_entity_ref: 'k:ns/other',
entity_id: 'my-id',
key: 'kind',
original_value: 'k',
value: 'k',
},
]);
const deferredStrategy = {
mode: 'deferred' as const,
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
};
await markForStitching({
knex,
strategy: deferredStrategy,
entityRefs: ['k:ns/n'],
});
await performStitching({
knex,
logger,
strategy: deferredStrategy,
entityRef: 'k:ns/n',
stitchTicket: (
await knex('stitch_queue')
.where('entity_ref', 'k:ns/n')
.select('stitch_ticket')
.first()
)?.stitch_ticket,
});
entities = await knex<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
entity = JSON.parse(entities[0].final_entity!);
expect(entity).toEqual({
relations: [
{
type: 'looksAt',
targetRef: 'k:ns/other',
},
],
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
etag: expect.any(String),
uid: 'my-id',
},
spec: {
k: 'v',
},
});
expect(entity.metadata.etag).toEqual(entities[0].hash);
const last_updated_at = entities[0].last_updated_at;
expect(last_updated_at).not.toBeNull();
const firstHash = entities[0].hash;
const search = await knex<DbSearchRow>('search');
expect(search).toEqual(
expect.arrayContaining([
{
entity_id: 'my-id',
key: 'relations.looksat',
original_value: 'k:ns/other',
value: 'k:ns/other',
},
{
entity_id: 'my-id',
key: 'apiversion',
original_value: 'a',
value: 'a',
},
{
entity_id: 'my-id',
key: 'kind',
original_value: 'k',
value: 'k',
},
{
entity_id: 'my-id',
key: 'metadata.name',
original_value: 'n',
value: 'n',
},
{
entity_id: 'my-id',
key: 'metadata.namespace',
original_value: 'ns',
value: 'ns',
},
{
entity_id: 'my-id',
key: 'metadata.uid',
original_value: 'my-id',
value: 'my-id',
},
{
entity_id: 'my-id',
key: 'spec.k',
original_value: 'v',
value: 'v',
},
]),
);
// Re-stitch without any changes
await markForStitching({
knex,
strategy: deferredStrategy,
entityRefs: ['k:ns/n'],
});
await performStitching({
knex,
logger,
strategy: deferredStrategy,
entityRef: 'k:ns/n',
stitchTicket: (
await knex('stitch_queue')
.where('entity_ref', 'k:ns/n')
.select('stitch_ticket')
.first()
)?.stitch_ticket,
});
entities = await knex<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
entity = JSON.parse(entities[0].final_entity!);
expect(entities[0].hash).toEqual(firstHash);
expect(entity.metadata.etag).toEqual(firstHash);
// Now add one more relation and re-stitch
await knex<DbRelationsRow>('relations').insert([
{
originating_entity_id: 'my-id',
source_entity_ref: 'k:ns/n',
type: 'looksAt',
target_entity_ref: 'k:ns/third',
entity_id: 'my-id',
key: 'metadata.name',
original_value: 'n',
value: 'n',
},
]);
await markForStitching({
knex,
strategy: deferredStrategy,
entityRefs: ['k:ns/n'],
});
await performStitching({
knex,
logger,
strategy: deferredStrategy,
entityRef: 'k:ns/n',
stitchTicket: (
await knex('stitch_queue')
.where('entity_ref', 'k:ns/n')
.select('stitch_ticket')
.first()
)?.stitch_ticket,
});
entities = await knex<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
entity = JSON.parse(entities[0].final_entity!);
expect(entity).toEqual({
relations: expect.arrayContaining([
{
type: 'looksAt',
targetRef: 'k:ns/other',
},
{
type: 'looksAt',
targetRef: 'k:ns/third',
},
]),
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
etag: expect.any(String),
uid: 'my-id',
{
entity_id: 'my-id',
key: 'metadata.namespace',
original_value: 'ns',
value: 'ns',
},
spec: {
k: 'v',
{
entity_id: 'my-id',
key: 'metadata.uid',
original_value: 'my-id',
value: 'my-id',
},
});
{
entity_id: 'my-id',
key: 'spec.k',
original_value: 'v',
value: 'v',
},
]),
);
expect(entities[0].hash).not.toEqual(firstHash);
expect(entities[0].hash).toEqual(entity.metadata.etag);
expect(await knex<DbSearchRow>('search')).toEqual(
expect.arrayContaining([
{
entity_id: 'my-id',
key: 'relations.looksat',
original_value: 'k:ns/other',
value: 'k:ns/other',
},
{
entity_id: 'my-id',
key: 'relations.looksat',
original_value: 'k:ns/third',
value: 'k:ns/third',
},
{
entity_id: 'my-id',
key: 'apiversion',
original_value: 'a',
value: 'a',
},
{
entity_id: 'my-id',
key: 'kind',
original_value: 'k',
value: 'k',
},
{
entity_id: 'my-id',
key: 'metadata.name',
original_value: 'n',
value: 'n',
},
{
entity_id: 'my-id',
key: 'metadata.namespace',
original_value: 'ns',
value: 'ns',
},
{
entity_id: 'my-id',
key: 'metadata.uid',
original_value: 'my-id',
value: 'my-id',
},
{
entity_id: 'my-id',
key: 'spec.k',
original_value: 'v',
value: 'v',
},
]),
);
// Re-stitch without any changes
await markForStitching({
knex,
entityRefs: ['k:ns/n'],
});
await performStitching({
knex,
logger,
entityRef: 'k:ns/n',
stitchTicket: (
await knex('stitch_queue')
.where('entity_ref', 'k:ns/n')
.select('stitch_ticket')
.first()
)?.stitch_ticket,
});
entities = await knex<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
entity = JSON.parse(entities[0].final_entity!);
expect(entities[0].hash).toEqual(firstHash);
expect(entity.metadata.etag).toEqual(firstHash);
// Now add one more relation and re-stitch
await knex<DbRelationsRow>('relations').insert([
{
originating_entity_id: 'my-id',
source_entity_ref: 'k:ns/n',
type: 'looksAt',
target_entity_ref: 'k:ns/third',
},
]);
await markForStitching({
knex,
entityRefs: ['k:ns/n'],
});
await performStitching({
knex,
logger,
entityRef: 'k:ns/n',
stitchTicket: (
await knex('stitch_queue')
.where('entity_ref', 'k:ns/n')
.select('stitch_ticket')
.first()
)?.stitch_ticket,
});
entities = await knex<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
entity = JSON.parse(entities[0].final_entity!);
expect(entity).toEqual({
relations: expect.arrayContaining([
{
type: 'looksAt',
targetRef: 'k:ns/other',
},
{
type: 'looksAt',
targetRef: 'k:ns/third',
},
]),
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
etag: expect.any(String),
uid: 'my-id',
},
spec: {
k: 'v',
},
});
expect(entities[0].hash).not.toEqual(firstHash);
expect(entities[0].hash).toEqual(entity.metadata.etag);
expect(await knex<DbSearchRow>('search')).toEqual(
expect.arrayContaining([
{
entity_id: 'my-id',
key: 'relations.looksat',
original_value: 'k:ns/other',
value: 'k:ns/other',
},
{
entity_id: 'my-id',
key: 'relations.looksat',
original_value: 'k:ns/third',
value: 'k:ns/third',
},
{
entity_id: 'my-id',
key: 'apiversion',
original_value: 'a',
value: 'a',
},
{
entity_id: 'my-id',
key: 'kind',
original_value: 'k',
value: 'k',
},
{
entity_id: 'my-id',
key: 'metadata.name',
original_value: 'n',
value: 'n',
},
{
entity_id: 'my-id',
key: 'metadata.namespace',
original_value: 'ns',
value: 'ns',
},
{
entity_id: 'my-id',
key: 'metadata.uid',
original_value: 'my-id',
value: 'my-id',
},
{
entity_id: 'my-id',
key: 'spec.k',
original_value: 'v',
value: 'v',
},
]),
);
},
);
describe.each(databases.eachSupportedId())(
'performStitching edge cases, %p',
databaseId => {
const logger = mockServices.logger.mock();
it('handles conflicts with past stitches', async () => {
if (databaseId === 'MYSQL_8') {
// MySQL doesn't handle conflicts in the same way as the other two, most
@@ -380,7 +371,6 @@ describe.each(databases.eachSupportedId())(
performStitching({
knex,
logger: stitchLogger,
strategy: { mode: 'immediate' },
entityRef: 'k:ns/n',
}),
).resolves.toBe('abandoned');
@@ -430,7 +420,6 @@ describe.each(databases.eachSupportedId())(
performStitching({
knex,
logger: stitchLogger,
strategy: { mode: 'immediate' },
entityRef: 'k:ns/n',
}),
).resolves.toBe('changed');
plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts
BIN
View File
Binary file not shown.
plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts
+4 -96
View File
@@ -16,7 +16,6 @@
import { TestDatabases } from '@backstage/backend-test-utils';
import { Knex } from 'knex';
import { StitchingStrategy } from '../../../stitching/types';
import { applyDatabaseMigrations } from '../../migrations';
import {
DbFinalEntitiesRow,
@@ -39,17 +38,14 @@ describe.each(databases.eachSupportedId())(
return knex;
}
async function run(
knex: Knex,
strategy: StitchingStrategy,
): Promise<number> {
async function run(knex: Knex): Promise<number> {
let result: number;
await knex.transaction(
async tx => {
// We can't return here, as knex swallows the return type in case the
// transaction is rolled back:
// https://github.com/knex/knex/blob/e37aeaa31c8ef9c1b07d2e4d3ec6607e557d800d/lib/transaction.js#L136
result = await deleteOrphanedEntities({ knex: tx, strategy });
result = await deleteOrphanedEntities({ knex: tx });
},
{
// If we explicitly trigger a rollback, don't fail.
@@ -133,7 +129,7 @@ describe.each(databases.eachSupportedId())(
});
}
it('works for some mixed paths in immediate mode', async () => {
it('works for some mixed paths', async () => {
/*
In this graph, edges represent refresh state references, not entity relations:
@@ -189,95 +185,7 @@ describe.each(databases.eachSupportedId())(
await insertRelation(knex, 'E2', 'E3');
await insertRelation(knex, 'E10', 'E6');
await insertRelation(knex, 'E7', 'E6');
await expect(run(knex, { mode: 'immediate' })).resolves.toEqual(5);
await expect(refreshState(knex)).resolves.toEqual([
{ entity_ref: 'E1', result_hash: 'original' },
{ entity_ref: 'E2', result_hash: 'force-stitching' },
{ entity_ref: 'E7', result_hash: 'force-stitching' },
{ entity_ref: 'E8', result_hash: 'original' },
{ entity_ref: 'E9', result_hash: 'original' },
]);
await expect(stitchQueue(knex)).resolves.toEqual([]);
await expect(finalEntities(knex)).resolves.toEqual([
{ entity_ref: 'E1', hash: 'original', next_stitch_at: null },
{
entity_ref: 'E2',
hash: 'force-stitching',
next_stitch_at: null,
},
{
entity_ref: 'E7',
hash: 'force-stitching',
next_stitch_at: null,
},
{ entity_ref: 'E8', hash: 'original', next_stitch_at: null },
{ entity_ref: 'E9', hash: 'original', next_stitch_at: null },
]);
});
it('works for some mixed paths in deferred mode', async () => {
/*
In this graph, edges represent refresh state references, not entity relations:
P1 - E1 -- E2
/
E3
/
E4
\
E5
/
E6
\
E7
/
P2 - E8
P3 - E9
E10
Result: E3, E4, E5, E6, and E10 deleted; others remain
Entities that had relations pointing at orphans are marked for reprocessing
*/
const knex = await createDatabase();
await insertEntity(
knex,
'E1',
'E2',
'E3',
'E4',
'E5',
'E6',
'E7',
'E8',
'E9',
'E10',
);
await insertReference(
knex,
{ source_key: 'P1', target_entity_ref: 'E1' },
{ source_entity_ref: 'E1', target_entity_ref: 'E2' },
{ source_entity_ref: 'E3', target_entity_ref: 'E2' },
{ source_entity_ref: 'E4', target_entity_ref: 'E3' },
{ source_entity_ref: 'E4', target_entity_ref: 'E5' },
{ source_entity_ref: 'E6', target_entity_ref: 'E5' },
{ source_entity_ref: 'E6', target_entity_ref: 'E7' },
{ source_key: 'P2', target_entity_ref: 'E8' },
{ source_entity_ref: 'E8', target_entity_ref: 'E7' },
{ source_key: 'P3', target_entity_ref: 'E9' },
);
await insertRelation(knex, 'E1', 'E2');
await insertRelation(knex, 'E2', 'E3');
await insertRelation(knex, 'E10', 'E6');
await insertRelation(knex, 'E7', 'E6');
await expect(
run(knex, {
mode: 'deferred',
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
}),
).resolves.toEqual(5);
await expect(run(knex)).resolves.toEqual(5);
await expect(refreshState(knex)).resolves.toEqual([
{ entity_ref: 'E1', result_hash: 'original' },
{ entity_ref: 'E2', result_hash: 'original' },
plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.ts
+1 -4
View File
@@ -16,7 +16,6 @@
import { Knex } from 'knex';
import uniq from 'lodash/uniq';
import { StitchingStrategy } from '../../../stitching/types';
import { DbRefreshStateRow } from '../../tables';
import { markForStitching } from '../stitcher/markForStitching';
@@ -27,9 +26,8 @@ import { markForStitching } from '../stitcher/markForStitching';
*/
export async function deleteOrphanedEntities(options: {
knex: Knex.Transaction | Knex;
strategy: StitchingStrategy;
}): Promise<number> {
const { knex, strategy } = options;
const { knex } = options;
let total = 0;
@@ -83,7 +81,6 @@ export async function deleteOrphanedEntities(options: {
// Mark all of the things that the orphans had relations to for stitching
await markForStitching({
knex,
strategy,
entityIds: orphanRelationIds,
});
}
plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts
+28 -25
View File
@@ -20,11 +20,17 @@ import waitForExpect from 'wait-for-expect';
import { DefaultProcessingDatabase } from '../database/DefaultProcessingDatabase';
import { DefaultCatalogProcessingEngine } from './DefaultCatalogProcessingEngine';
import { CatalogProcessingOrchestrator } from './types';
import { Stitcher } from '../stitching/types';
import { ConfigReader } from '@backstage/config';
import { mockServices } from '@backstage/backend-test-utils';
import { metricsServiceMock } from '@backstage/backend-test-utils/alpha';
jest.mock('../database/operations/stitcher/markForStitching', () => ({
markForStitching: jest.fn().mockResolvedValue(undefined),
}));
const { markForStitching } = jest.requireMock(
'../database/operations/stitcher/markForStitching',
) as { markForStitching: jest.Mock };
describe('DefaultCatalogProcessingEngine', () => {
const db = {
transaction: jest.fn(),
@@ -37,9 +43,6 @@ describe('DefaultCatalogProcessingEngine', () => {
const orchestrator: jest.Mocked<CatalogProcessingOrchestrator> = {
process: jest.fn(),
};
const stitcher = {
stitch: jest.fn(),
} as unknown as jest.Mocked<Stitcher>;
const hash = {
update: () => hash,
digest: jest.fn(),
@@ -69,7 +72,7 @@ describe('DefaultCatalogProcessingEngine', () => {
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
createHash: () => hash,
scheduler: mockServices.scheduler(),
events: mockServices.events.mock(),
@@ -139,7 +142,7 @@ describe('DefaultCatalogProcessingEngine', () => {
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
scheduler: mockServices.scheduler(),
createHash: () => hash,
events: mockServices.events.mock(),
@@ -225,7 +228,7 @@ describe('DefaultCatalogProcessingEngine', () => {
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
scheduler: mockServices.scheduler(),
createHash: () => hash,
events: mockServices.events.mock(),
@@ -305,7 +308,7 @@ describe('DefaultCatalogProcessingEngine', () => {
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
scheduler: mockServices.scheduler(),
createHash: () => hash,
events: mockServices.events.mock(),
@@ -367,7 +370,7 @@ describe('DefaultCatalogProcessingEngine', () => {
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
scheduler: mockServices.scheduler(),
createHash: () => hash,
pollingIntervalMs: 100,
@@ -467,12 +470,12 @@ describe('DefaultCatalogProcessingEngine', () => {
await engine.start();
await waitForExpect(() => {
expect(stitcher.stitch).toHaveBeenCalledTimes(2);
expect(markForStitching).toHaveBeenCalledTimes(2);
});
expect([...stitcher.stitch.mock.calls[0][0].entityRefs!]).toEqual(
expect([...markForStitching.mock.calls[0][0].entityRefs!]).toEqual(
expect.arrayContaining(['k:ns/me', 'k:ns/other1', 'k:ns/other2']),
);
expect([...stitcher.stitch.mock.calls[1][0].entityRefs!]).toEqual(
expect([...markForStitching.mock.calls[1][0].entityRefs!]).toEqual(
expect.arrayContaining(['k:ns/me', 'k:ns/other1', 'k:ns/other3']),
);
await engine.stop();
@@ -485,7 +488,7 @@ describe('DefaultCatalogProcessingEngine', () => {
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
scheduler: mockServices.scheduler(),
createHash: () => hash,
pollingIntervalMs: 100,
@@ -572,15 +575,15 @@ describe('DefaultCatalogProcessingEngine', () => {
await engine.start();
await waitForExpect(() => {
expect(stitcher.stitch).toHaveBeenCalledTimes(2);
expect(markForStitching).toHaveBeenCalledTimes(2);
});
expect([...stitcher.stitch.mock.calls[0][0].entityRefs!]).toEqual(
expect([...markForStitching.mock.calls[0][0].entityRefs!]).toEqual(
expect.arrayContaining(['k:ns/me', 'k:ns/other1']),
);
// As a result of switching the relationship for source other1 to
// a new target entity, the other1 relationship source must be
// restitched.
expect([...stitcher.stitch.mock.calls[1][0].entityRefs!]).toEqual(
expect([...markForStitching.mock.calls[1][0].entityRefs!]).toEqual(
expect.arrayContaining(['k:ns/me', 'k:ns/other1']),
);
await engine.stop();
@@ -593,7 +596,7 @@ describe('DefaultCatalogProcessingEngine', () => {
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
scheduler: mockServices.scheduler(),
createHash: () => hash,
pollingIntervalMs: 100,
@@ -664,9 +667,9 @@ describe('DefaultCatalogProcessingEngine', () => {
await engine.start();
await waitForExpect(() => {
expect(stitcher.stitch).toHaveBeenCalledTimes(1);
expect(markForStitching).toHaveBeenCalledTimes(1);
});
expect([...stitcher.stitch.mock.calls[0][0].entityRefs!]).toEqual(
expect([...markForStitching.mock.calls[0][0].entityRefs!]).toEqual(
expect.arrayContaining(['k:ns/me']),
);
await engine.stop();
@@ -679,7 +682,7 @@ describe('DefaultCatalogProcessingEngine', () => {
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
scheduler: mockServices.scheduler(),
createHash: () => hash,
pollingIntervalMs: 100,
@@ -755,9 +758,9 @@ describe('DefaultCatalogProcessingEngine', () => {
await engine.start();
await waitForExpect(() => {
expect(stitcher.stitch).toHaveBeenCalledTimes(1);
expect(markForStitching).toHaveBeenCalledTimes(1);
});
expect([...stitcher.stitch.mock.calls[0][0].entityRefs!]).toEqual(
expect([...markForStitching.mock.calls[0][0].entityRefs!]).toEqual(
expect.arrayContaining(['k:ns/me', 'k:ns/other2']),
);
await engine.stop();
@@ -770,7 +773,7 @@ describe('DefaultCatalogProcessingEngine', () => {
processingDatabase: db,
knex: {} as any,
orchestrator: orchestrator,
stitcher: stitcher,
scheduler: mockServices.scheduler(),
createHash: () => hash,
pollingIntervalMs: 100,
@@ -836,9 +839,9 @@ describe('DefaultCatalogProcessingEngine', () => {
await engine.start();
await waitForExpect(() => {
expect(stitcher.stitch).toHaveBeenCalledTimes(1);
expect(markForStitching).toHaveBeenCalledTimes(1);
});
expect([...stitcher.stitch.mock.calls[0][0].entityRefs!]).toEqual(
expect([...markForStitching.mock.calls[0][0].entityRefs!]).toEqual(
expect.arrayContaining(['k:ns/me', 'k:ns/other2']),
);
await engine.stop();
plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts
+5 -11
View File
@@ -27,7 +27,7 @@ import { trace } from '@opentelemetry/api';
import { ProcessingDatabase, RefreshStateItem } from '../database/types';
import { createCounterMetric, createSummaryMetric } from '../util/metrics';
import { CatalogProcessingOrchestrator, EntityProcessingResult } from './types';
import { Stitcher, stitchingStrategyFromConfig } from '../stitching/types';
import { markForStitching } from '../database/operations/stitcher/markForStitching';
import { startTaskPipeline } from './TaskPipeline';
import { Config } from '@backstage/config';
import {
@@ -65,7 +65,6 @@ export class DefaultCatalogProcessingEngine {
private readonly knex: Knex;
private readonly processingDatabase: ProcessingDatabase;
private readonly orchestrator: CatalogProcessingOrchestrator;
private readonly stitcher: Stitcher;
private readonly createHash: () => Hash;
private readonly pollingIntervalMs: number;
private readonly orphanCleanupIntervalMs: number;
@@ -85,7 +84,6 @@ export class DefaultCatalogProcessingEngine {
knex: Knex;
processingDatabase: ProcessingDatabase;
orchestrator: CatalogProcessingOrchestrator;
stitcher: Stitcher;
createHash: () => Hash;
pollingIntervalMs?: number;
orphanCleanupIntervalMs?: number;
@@ -103,7 +101,6 @@ export class DefaultCatalogProcessingEngine {
this.knex = options.knex;
this.processingDatabase = options.processingDatabase;
this.orchestrator = options.orchestrator;
this.stitcher = options.stitcher;
this.createHash = options.createHash;
this.pollingIntervalMs = options.pollingIntervalMs ?? 1_000;
this.orphanCleanupIntervalMs = options.orphanCleanupIntervalMs ?? 30_000;
@@ -283,7 +280,8 @@ export class DefaultCatalogProcessingEngine {
});
});
await this.stitcher.stitch({
await markForStitching({
knex: this.knex,
entityRefs: [stringifyEntityRef(unprocessedEntity)],
});
@@ -338,7 +336,8 @@ export class DefaultCatalogProcessingEngine {
}
});
await this.stitcher.stitch({
await markForStitching({
knex: this.knex,
entityRefs: setOfThingsToStitch,
});
@@ -358,15 +357,10 @@ export class DefaultCatalogProcessingEngine {
return () => {};
}
const stitchingStrategy = stitchingStrategyFromConfig(this.config, {
logger: this.logger,
});
const runOnce = async () => {
try {
const n = await deleteOrphanedEntities({
knex: this.knex,
strategy: stitchingStrategy,
});
if (n > 0) {
this.logger.info(`Deleted ${n} orphaned entities`);
plugins/catalog-backend/src/service/CatalogBuilder.ts
-2
View File
@@ -435,7 +435,6 @@ export class CatalogBuilder {
const unauthorizedEntitiesCatalog = new DefaultEntitiesCatalog({
database: dbClient,
logger,
stitcher,
enableRelationsCompatibility,
});
@@ -511,7 +510,6 @@ export class CatalogBuilder {
knex: dbClient,
processingDatabase,
orchestrator,
stitcher,
createHash: () => createHash('sha1'),
pollingIntervalMs: 1000,
onProcessingError: event => {
plugins/catalog-backend/src/service/DefaultEntitiesCatalog.test.ts
+11 -46
View File
@@ -33,7 +33,6 @@ import {
DbRefreshStateRow,
DbSearchRow,
} from '../database/tables';
import { Stitcher } from '../stitching/types';
import { DefaultEntitiesCatalog } from './DefaultEntitiesCatalog';
import { EntitiesRequest } from '../catalog/types';
import { buildEntitySearch } from '../database/operations/stitcher/buildEntitySearch';
@@ -52,9 +51,6 @@ describe.each(databases.eachSupportedId())(
await knex.destroy();
});
const stitch = jest.fn();
const stitcher: Stitcher = { stitch } as any;
async function createDatabase() {
knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
@@ -165,7 +161,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const result = await catalog.entityAncestry('k:default/root');
expect(result.rootEntityRef).toEqual('k:default/root');
@@ -195,7 +190,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
await expect(() =>
catalog.entityAncestry('k:default/root'),
@@ -238,7 +232,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const result = await catalog.entityAncestry('k:default/root');
expect(result.rootEntityRef).toEqual('k:default/root');
@@ -294,7 +287,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const testFilter = {
@@ -331,7 +323,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const testFilter = {
@@ -382,7 +373,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const testFilter1 = {
@@ -439,7 +429,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const testFilter1 = {
@@ -484,7 +473,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const testFilter = {
@@ -530,7 +518,7 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
enableRelationsCompatibility: true,
});
@@ -585,7 +573,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
function f(
@@ -648,7 +635,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
function f(
@@ -726,7 +712,7 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
async function page(limit: number, offset?: number): Promise<string[]> {
@@ -795,7 +781,7 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
async function page(
@@ -862,7 +848,7 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
async function page(order: 'asc' | 'desc'): Promise<string[]> {
@@ -910,7 +896,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const res = await catalog.entitiesBatch({
@@ -963,7 +948,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const res = await catalog.entitiesBatch({
@@ -1009,7 +993,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const filter = {
@@ -1183,7 +1166,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const filter = {
@@ -1358,7 +1340,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const filter = {
@@ -1415,7 +1396,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const request: QueryEntitiesInitialRequest = {
@@ -1466,7 +1446,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const filter = {
@@ -1562,7 +1541,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const filter = {
@@ -1639,7 +1617,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const request: QueryEntitiesInitialRequest = {
@@ -1672,7 +1649,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const request: QueryEntitiesInitialRequest = {
@@ -1720,7 +1696,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const limit = 2;
@@ -1841,7 +1816,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const limit = 2;
@@ -1902,7 +1876,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
const limit = 2;
@@ -1994,7 +1967,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
// Entities without the sort field are excluded — sorting by a field
@@ -2034,7 +2006,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
await expect(
@@ -2130,7 +2101,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
// Query with orderField
@@ -2161,7 +2131,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
// Use filter to restrict to kind=component, and query to restrict to name=A
@@ -2246,7 +2215,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
await catalog.removeEntityByUid(uid);
@@ -2262,9 +2230,13 @@ describe.each(databases.eachSupportedId())(
{ entity_ref: 'k:default/unrelated1', result_hash: 'not-changed' },
{ entity_ref: 'k:default/unrelated2', result_hash: 'not-changed' },
]);
expect(stitch).toHaveBeenCalledWith({
entityRefs: new Set(['k:default/unrelated1', 'k:default/unrelated2']),
});
const stitchQueue = await knex('stitch_queue')
.select('entity_ref')
.orderBy('entity_ref');
expect(stitchQueue.map(r => r.entity_ref)).toEqual([
'k:default/unrelated1',
'k:default/unrelated2',
]);
});
});
@@ -2293,7 +2265,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
await expect(
@@ -2364,7 +2335,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
await expect(
@@ -2410,7 +2380,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
await expect(
@@ -2451,7 +2420,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
await expect(
@@ -2500,7 +2468,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
await expect(
@@ -2523,7 +2490,6 @@ describe.each(databases.eachSupportedId())(
return new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
}
@@ -2571,7 +2537,6 @@ describe.each(databases.eachSupportedId())(
const catalog = new DefaultEntitiesCatalog({
database: knex,
logger: mockServices.logger.mock(),
stitcher,
});
// With filter: unstitched entity should be excluded because the
plugins/catalog-backend/src/service/DefaultEntitiesCatalog.ts
+3 -5
View File
@@ -41,7 +41,7 @@ import {
DbRelationsRow,
DbSearchRow,
} from '../database/tables';
import { Stitcher } from '../stitching/types';
import { markForStitching } from '../database/operations/stitcher/markForStitching';
import {
expandLegacyCompoundRelationsInEntity,
@@ -102,18 +102,15 @@ function stringifyPagination(
export class DefaultEntitiesCatalog implements EntitiesCatalog {
private readonly database: Knex;
private readonly logger: LoggerService;
private readonly stitcher: Stitcher;
private readonly enableRelationsCompatibility: boolean;
constructor(options: {
database: Knex;
logger: LoggerService;
stitcher: Stitcher;
enableRelationsCompatibility?: boolean;
}) {
this.database = options.database;
this.logger = options.logger;
this.stitcher = options.stitcher;
this.enableRelationsCompatibility = Boolean(
options.enableRelationsCompatibility,
);
@@ -759,7 +756,8 @@ export class DefaultEntitiesCatalog implements EntitiesCatalog {
});
if (relationPeerRefs.size > 0) {
await this.stitcher.stitch({
await markForStitching({
knex: this.database,
entityRefs: relationPeerRefs,
});
}
plugins/catalog-backend/src/service/DefaultRefreshService.test.ts
-7
View File
@@ -35,7 +35,6 @@ import { DefaultCatalogProcessingEngine } from '../processing/DefaultCatalogProc
import { EntityProcessingRequest } from '../processing/types';
import { DefaultRefreshService } from './DefaultRefreshService';
import { ConfigReader } from '@backstage/config';
import { DefaultStitcher } from '../stitching/DefaultStitcher';
import { LoggerService } from '@backstage/backend-plugin-api';
import { metricsServiceMock } from '@backstage/backend-test-utils/alpha';
@@ -113,17 +112,11 @@ describe.each(databases.eachSupportedId())(
}
}
const stitcher = DefaultStitcher.fromConfig(new ConfigReader({}), {
knex,
logger: defaultLogger,
metrics: metricsServiceMock.mock(),
});
const engine = new DefaultCatalogProcessingEngine({
config: new ConfigReader({}),
logger: defaultLogger,
processingDatabase: db,
knex: knex,
stitcher: stitcher,
scheduler: mockServices.scheduler(),
orchestrator: {
async process(request: EntityProcessingRequest) {
plugins/catalog-backend/src/stitching/DefaultStitcher.test.ts
+50 -8
View File
@@ -17,6 +17,7 @@
import { TestDatabases, mockServices } from '@backstage/backend-test-utils';
import { Entity } from '@backstage/catalog-model';
import { applyDatabaseMigrations } from '../database/migrations';
import { markForStitching } from '../database/operations/stitcher/markForStitching';
import {
DbFinalEntitiesRow,
DbRefreshStateReferencesRow,
@@ -41,11 +42,12 @@ describe.each(databases.eachSupportedId())('Stitcher, %p', databaseId => {
const stitcher = new DefaultStitcher({
knex: db,
logger,
strategy: { mode: 'immediate' },
strategy: {
pollingInterval: { milliseconds: 50 },
stitchTimeout: { seconds: 10 },
},
metrics: metricsServiceMock.mock(),
});
let entities: DbFinalEntitiesRow[];
let entity: Entity;
await db<DbRefreshStateRow>('refresh_state').insert([
{
@@ -87,12 +89,21 @@ describe.each(databases.eachSupportedId())('Stitcher, %p', databaseId => {
},
]);
await stitcher.stitch({ entityRefs: ['k:ns/n'] });
await markForStitching({ knex: db, entityRefs: ['k:ns/n'] });
await stitcher.start();
entities = await db<DbFinalEntitiesRow>('final_entities');
// Wait for stitching to complete
await waitForCondition(async () => {
const entities = await db<DbFinalEntitiesRow>('final_entities');
return entities.length === 1 && entities[0].final_entity !== null;
});
await stitcher.stop();
let entities = await db<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
entity = JSON.parse(entities[0].final_entity!);
let entity: Entity = JSON.parse(entities[0].final_entity!);
expect(entity).toEqual({
relations: [
{
@@ -167,7 +178,15 @@ describe.each(databases.eachSupportedId())('Stitcher, %p', databaseId => {
);
// Re-stitch without any changes
await stitcher.stitch({ entityRefs: ['k:ns/n'] });
await markForStitching({ knex: db, entityRefs: ['k:ns/n'] });
await stitcher.start();
await waitForCondition(async () => {
const queue = await db('stitch_queue').select('*');
return queue.length === 0;
});
await stitcher.stop();
entities = await db<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
@@ -185,7 +204,15 @@ describe.each(databases.eachSupportedId())('Stitcher, %p', databaseId => {
},
]);
await stitcher.stitch({ entityRefs: ['k:ns/n'] });
await markForStitching({ knex: db, entityRefs: ['k:ns/n'] });
await stitcher.start();
await waitForCondition(async () => {
const e = await db<DbFinalEntitiesRow>('final_entities');
return e.length === 1 && e[0].hash !== firstHash;
});
await stitcher.stop();
entities = await db<DbFinalEntitiesRow>('final_entities');
@@ -272,3 +299,18 @@ describe.each(databases.eachSupportedId())('Stitcher, %p', databaseId => {
);
});
});
async function waitForCondition(
condition: () => Promise<boolean>,
timeoutMs = 10_000,
intervalMs = 50,
) {
const start = Date.now();
while (Date.now() - start < timeoutMs) {
if (await condition()) {
return;
}
await new Promise(resolve => setTimeout(resolve, intervalMs));
}
throw new Error('Timed out waiting for condition');
}
plugins/catalog-backend/src/stitching/DefaultStitcher.ts
+29 -79
View File
@@ -17,19 +17,12 @@
import { Config } from '@backstage/config';
import { durationToMilliseconds, HumanDuration } from '@backstage/types';
import { Knex } from 'knex';
import splitToChunks from 'lodash/chunk';
import { DateTime } from 'luxon';
import { getDeferredStitchableEntities } from '../database/operations/stitcher/getDeferredStitchableEntities';
import { markForStitching } from '../database/operations/stitcher/markForStitching';
import { performStitching } from '../database/operations/stitcher/performStitching';
import { DbRefreshStateRow } from '../database/tables';
import { startTaskPipeline } from '../processing/TaskPipeline';
import { progressTracker } from './progressTracker';
import {
Stitcher,
StitchingStrategy,
stitchingStrategyFromConfig,
} from './types';
import { StitchingStrategy, stitchingStrategyFromConfig } from './types';
import { LoggerService } from '@backstage/backend-plugin-api';
import { MetricsService } from '@backstage/backend-plugin-api/alpha';
@@ -44,7 +37,7 @@ type StitchProgressTracker = ReturnType<typeof progressTracker>;
* ingestion process, and stitching them together into the final entity JSON
* shape.
*/
export class DefaultStitcher implements Stitcher {
export class DefaultStitcher {
private readonly knex: Knex;
private readonly logger: LoggerService;
private readonly strategy: StitchingStrategy;
@@ -85,80 +78,38 @@ export class DefaultStitcher implements Stitcher {
);
}
async stitch(options: {
entityRefs?: Iterable<string>;
entityIds?: Iterable<string>;
}) {
const { entityRefs, entityIds } = options;
if (this.strategy.mode === 'deferred') {
await markForStitching({
knex: this.knex,
strategy: this.strategy,
entityRefs,
entityIds,
});
return;
}
if (entityRefs) {
for (const entityRef of entityRefs) {
await this.#stitchOne({ entityRef });
}
}
if (entityIds) {
const chunks = splitToChunks(
Array.isArray(entityIds) ? entityIds : [...entityIds],
100,
);
for (const chunk of chunks) {
const rows = await this.knex<DbRefreshStateRow>('refresh_state')
.select('entity_ref')
.whereIn('entity_id', chunk);
for (const row of rows) {
await this.#stitchOne({ entityRef: row.entity_ref });
}
}
}
}
async start() {
if (this.strategy.mode === 'deferred') {
if (this.stopFunc) {
throw new Error('Processing engine is already started');
}
const { pollingInterval, stitchTimeout } = this.strategy;
const stopPipeline = startTaskPipeline<DeferredStitchItem>({
lowWatermark: 2,
highWatermark: 5,
pollingIntervalMs: durationToMilliseconds(pollingInterval),
loadTasks: async count => {
return await this.#getStitchableEntities(count, stitchTimeout);
},
processTask: async item => {
return await this.#stitchOne({
entityRef: item.entityRef,
stitchTicket: item.stitchTicket,
stitchRequestedAt: item.stitchRequestedAt,
});
},
});
this.stopFunc = () => {
stopPipeline();
};
if (this.stopFunc) {
throw new Error('Processing engine is already started');
}
const { pollingInterval, stitchTimeout } = this.strategy;
const stopPipeline = startTaskPipeline<DeferredStitchItem>({
lowWatermark: 2,
highWatermark: 5,
pollingIntervalMs: durationToMilliseconds(pollingInterval),
loadTasks: async count => {
return await this.#getStitchableEntities(count, stitchTimeout);
},
processTask: async item => {
return await this.#stitchOne({
entityRef: item.entityRef,
stitchTicket: item.stitchTicket,
stitchRequestedAt: item.stitchRequestedAt,
});
},
});
this.stopFunc = () => {
stopPipeline();
};
}
async stop() {
if (this.strategy.mode === 'deferred') {
if (this.stopFunc) {
this.stopFunc();
this.stopFunc = undefined;
}
if (this.stopFunc) {
this.stopFunc();
this.stopFunc = undefined;
}
}
@@ -189,7 +140,6 @@ export class DefaultStitcher implements Stitcher {
const result = await performStitching({
knex: this.knex,
logger: this.logger,
strategy: this.strategy,
entityRef: options.entityRef,
stitchTicket: options.stitchTicket,
});
plugins/catalog-backend/src/stitching/types.ts
+24 -69
View File
@@ -19,85 +19,40 @@ import { Config, readDurationFromConfig } from '@backstage/config';
import { HumanDuration } from '@backstage/types';
/**
* Performs the act of stitching - to take all of the various outputs from the
* ingestion process, and stitching them together into the final entity JSON
* shape.
* Configuration for the stitching process, controlling polling and timeout
* behavior for the deferred stitching worker.
*/
export interface Stitcher {
stitch(options: {
entityRefs?: Iterable<string>;
entityIds?: Iterable<string>;
}): Promise<void>;
}
/**
* The strategies supported by the stitching process, in terms of when to
* perform stitching.
*
* @remarks
*
* In immediate mode, stitching happens "in-band" (blocking) immediately when
* each processing task finishes. When set to `'deferred'`, stitching is instead
* deferred to happen on a separate asynchronous worker queue just like
* processing.
*
* Deferred stitching should make performance smoother when ingesting large
* amounts of entities, and reduce p99 processing times and repeated
* over-stitching of hot spot entities when fan-out/fan-in in terms of relations
* is very large. It does however also come with some performance cost due to
* the queuing with how much wall-clock time some types of task take.
*
* Note: Immediate mode is deprecated and will be removed in a future release.
*/
export type StitchingStrategy =
| {
mode: 'immediate';
}
| {
mode: 'deferred';
pollingInterval: HumanDuration;
stitchTimeout: HumanDuration;
};
let immediateDeprecationLogged = false;
export type StitchingStrategy = {
pollingInterval: HumanDuration;
stitchTimeout: HumanDuration;
};
export function stitchingStrategyFromConfig(
config: Config,
options: { logger: LoggerService },
options?: { logger?: LoggerService },
): StitchingStrategy {
const strategyMode = config.getOptionalString(
'catalog.stitchingStrategy.mode',
);
if (strategyMode === 'immediate') {
if (!immediateDeprecationLogged) {
immediateDeprecationLogged = true;
options.logger.warn(
'DEPRECATED: Immediate mode stitching has been deprecated, and will be removed in the next Backstage release.',
);
}
return {
mode: 'immediate',
};
} else if (strategyMode === undefined || strategyMode === 'deferred') {
const pollingIntervalKey = 'catalog.stitchingStrategy.pollingInterval';
const stitchTimeoutKey = 'catalog.stitchingStrategy.stitchTimeout';
const pollingInterval = config.has(pollingIntervalKey)
? readDurationFromConfig(config, { key: pollingIntervalKey })
: { seconds: 1 };
const stitchTimeout = config.has(stitchTimeoutKey)
? readDurationFromConfig(config, { key: stitchTimeoutKey })
: { seconds: 60 };
return {
mode: 'deferred',
pollingInterval: pollingInterval,
stitchTimeout: stitchTimeout,
};
options?.logger?.warn(
"The 'immediate' stitching strategy mode has been removed and is no longer supported. Falling back to deferred stitching. Please remove the 'catalog.stitchingStrategy.mode' configuration key.",
);
}
throw new Error(
`Invalid stitching strategy mode '${strategyMode}', expected one of 'immediate' or 'deferred'`,
);
const pollingIntervalKey = 'catalog.stitchingStrategy.pollingInterval';
const stitchTimeoutKey = 'catalog.stitchingStrategy.stitchTimeout';
const pollingInterval = config.has(pollingIntervalKey)
? readDurationFromConfig(config, { key: pollingIntervalKey })
: { seconds: 1 };
const stitchTimeout = config.has(stitchTimeoutKey)
? readDurationFromConfig(config, { key: stitchTimeoutKey })
: { seconds: 60 };
return {
pollingInterval: pollingInterval,
stitchTimeout: stitchTimeout,
};
}
plugins/catalog-backend/src/tests/integration.test.ts
+17 -18
View File
@@ -202,6 +202,7 @@ class WaitingProgressTracker implements ProgressTrackerWithErrorReports {
class TestHarness {
readonly #catalog: EntitiesCatalog;
readonly #engine: CatalogProcessingEngine;
readonly #stitcher: DefaultStitcher;
readonly #refresh: RefreshService;
readonly #provider: TestProvider;
readonly #proxyProgressTracker: ProxyProgressTracker;
@@ -226,11 +227,6 @@ class TestHarness {
connection: ':memory:',
},
},
catalog: {
stitchingStrategy: {
mode: 'immediate',
},
},
});
const logger = options?.logger ?? mockServices.logger.mock();
@@ -287,7 +283,6 @@ class TestHarness {
const catalog = new DefaultEntitiesCatalog({
database: options.db,
logger,
stitcher,
enableRelationsCompatibility: options?.enableRelationsCompatibility,
});
const proxyProgressTracker = new ProxyProgressTracker(
@@ -300,7 +295,6 @@ class TestHarness {
processingDatabase,
knex: options.db,
orchestrator,
stitcher,
scheduler: mockServices.scheduler(),
createHash: () => createHash('sha1'),
pollingIntervalMs: 50,
@@ -335,16 +329,8 @@ class TestHarness {
return new TestHarness(
catalog,
{
async start() {
await engine.start();
await stitcher.start();
},
async stop() {
await engine.stop();
await stitcher.stop();
},
},
engine,
stitcher,
refresh,
provider,
proxyProgressTracker,
@@ -355,6 +341,7 @@ class TestHarness {
constructor(
catalog: EntitiesCatalog,
engine: CatalogProcessingEngine,
stitcher: DefaultStitcher,
refresh: RefreshService,
provider: TestProvider,
proxyProgressTracker: ProxyProgressTracker,
@@ -362,6 +349,7 @@ class TestHarness {
) {
this.#catalog = catalog;
this.#engine = engine;
this.#stitcher = stitcher;
this.#refresh = refresh;
this.#provider = provider;
this.#proxyProgressTracker = proxyProgressTracker;
@@ -373,12 +361,24 @@ class TestHarness {
this.#proxyProgressTracker.setTracker(tracker);
this.#engine.start();
await this.#stitcher.start();
const errors = await tracker.wait();
this.#engine.stop();
await tracker.waitForFinish();
// Wait for the stitch queue to drain while the stitcher is still running
const start = Date.now();
while (Date.now() - start < 10_000) {
const queue = await this.#db('stitch_queue').select('*');
if (queue.length === 0) {
break;
}
await new Promise(resolve => setTimeout(resolve, 50));
}
await this.#stitcher.stop();
this.#proxyProgressTracker.setTracker(new NoopProgressTracker());
return errors;
@@ -411,7 +411,6 @@ class TestHarness {
async removeOrphanedEntities() {
await deleteOrphanedEntities({
knex: this.#db,
strategy: { mode: 'immediate' },
});
}
plugins/catalog-backend/src/tests/performance/stitchingPerformance.test.ts
+5 -55
View File
@@ -23,7 +23,6 @@ import {
import { catalogProcessingExtensionPoint } from '@backstage/plugin-catalog-node';
import { createDeferred } from '@backstage/types';
import { Knex } from 'knex';
import { default as catalogPlugin } from '../..';
import { applyDatabaseMigrations } from '../../database/migrations';
import {
SyntheticLoadEntitiesProcessor,
@@ -149,58 +148,9 @@ const databases = TestDatabases.create({
});
describePerformanceTest('stitchingPerformance', () => {
describe.each(databases.eachSupportedId())('%p', databaseId => {
it('runs stitching in immediate mode', async () => {
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
const load: SyntheticLoadOptions = {
baseEntitiesCount: 1000,
baseRelationsCount: 3,
baseRelationsSkew: 0.3,
childrenCount: 3,
};
const config = {
backend: { baseUrl: 'http://localhost:7007' },
catalog: { stitchingStrategy: { mode: 'immediate' } },
};
const tracker = new Tracker(knex, load);
const backend = await startTestBackend({
features: [
catalogPlugin,
mockServices.rootConfig.factory({ data: config }),
mockServices.database.factory({ knex }),
createBackendModule({
pluginId: 'catalog',
moduleId: 'synthetic-load-entities',
register(reg) {
reg.registerInit({
deps: {
catalog: catalogProcessingExtensionPoint,
},
async init({ catalog }) {
catalog.addEntityProvider(
new SyntheticLoadEntitiesProvider(load, tracker.events()),
);
catalog.addProcessor(
new SyntheticLoadEntitiesProcessor(load),
);
},
});
},
}),
],
});
await expect(tracker.completion()).resolves.toBeUndefined();
await backend.stop();
await knex.destroy();
});
it('runs stitching in deferred mode', async () => {
it.each(databases.eachSupportedId())(
'runs stitching in deferred mode, %p',
async databaseId => {
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
@@ -248,6 +198,6 @@ describePerformanceTest('stitchingPerformance', () => {
await expect(tracker.completion()).resolves.toBeUndefined();
await backend.stop();
await knex.destroy();
});
});
},
);
});