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

Merge pull request #18062 from backstage/freben/hack-stitch-queue

Browse Source 
introduce the concept of deferred stitching
This commit is contained in:
Patrik Oldsberg
2023-10-24 12:21:12 +02:00
committed by GitHub
parent 664e86f7f9 2bc82320e2
commit 0e6b506b3a
20 changed files with 1170 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/loud-ghosts-deny.md
+16
View File
@@ -0,0 +1,16 @@
---
'@backstage/plugin-catalog-backend': minor
---
Introduce a new optional config parameter `catalog.stitchingStrategy.mode`,
which can have the values `'immediate'` (default) and `'deferred'`. The default
is for stitching to work as it did before this change, which means that it
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.
plugins/catalog-backend/config.d.ts
Vendored
+13
View File
@@ -145,6 +145,19 @@ export interface Config {
*/
orphanStrategy?: 'keep' | 'delete';
/**
* The strategy to use when stitching together the final entities.
*/
stitchingStrategy?:
| {
/** Perform stitching in-band immediately when needed */
mode: 'immediate';
}
| {
/** Defer stitching to be performed asynchronously */
mode: 'deferred';
};
/**
* The interval at which the catalog should process its entities.
*
plugins/catalog-backend/migrations/20230525141717_stitch_queue.js
+85
View File
@@ -0,0 +1,85 @@
/*
* Copyright 2022 The Backstage Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
// @ts-check
/**
* @param { import("knex").Knex } knex
*/
exports.up = async function up(knex) {
await knex.schema.alterTable('refresh_state', table => {
table
.dateTime('next_stitch_at')
.nullable()
.comment('Timestamp of when entity should be stitched');
table
.string('next_stitch_ticket')
.nullable()
.comment('Random value distinguishing stitch requests');
table.index('next_stitch_at', 'refresh_state_next_stitch_at_idx', {
predicate: knex.whereNotNull('next_stitch_at'),
});
});
// Look for things that are due for stitching, and move them over to the new
// system. An explanation on the length based ones: Before adding the new
// system, stitching was triggered by setting hashes to various hard coded
// strings - and we leverage the fact that those strings were always shorter
// than a real hash would be. There's also the case where we have not yet
// created the final_entities row corresponding to the refresh_state one. This
// is split into two statements because MySQL doesn't allow you to write to
// tables you're reading from.
const candidates = await knex
.select({ entityId: 'refresh_state.entity_id' })
.from('refresh_state')
.leftOuterJoin(
'final_entities',
'final_entities.entity_id',
'refresh_state.entity_id',
)
// no final entities at all
.orWhereNull('final_entities.entity_id')
// the final entity hash was forcibly set to something short
.orWhere(knex.raw('LENGTH(??) < 15', ['final_entities.hash']))
// there used to be an ongoing stitch (possibly unfinished and aborted)
.orWhere(knex.raw('LENGTH(??) > 0', ['final_entities.stitch_ticket']))
// the processing output entity hash was forcibly set to something short
.orWhere(knex.raw('LENGTH(??) < 15', ['refresh_state.result_hash']));
if (candidates.length) {
for (let i = 0; i < candidates.length; i += 1000) {
await knex('refresh_state')
.update({
next_stitch_at: knex.fn.now(),
next_stitch_ticket: 'initial',
})
.whereIn(
'entity_id',
candidates.slice(i, i + 1000).map(c => c.entityId),
);
}
}
};
/**
* @param { import("knex").Knex } knex
*/
exports.down = async function down(knex) {
await knex.schema.alterTable('refresh_state', table => {
table.dropIndex([], 'refresh_state_next_stitch_at_idx');
table.dropColumn('next_stitch_at');
table.dropColumn('next_stitch_ticket');
});
};
plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.test.ts
+114
View File
@@ -0,0 +1,114 @@
/*
* Copyright 2023 The Backstage Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import { TestDatabases } from '@backstage/backend-test-utils';
import { applyDatabaseMigrations } from '../../migrations';
import { getDeferredStitchableEntities } from './getDeferredStitchableEntities';
jest.setTimeout(60_000);
describe('getDeferredStitchableEntities', () => {
const databases = TestDatabases.create({
ids: ['MYSQL_8', 'POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'],
});
it.each(databases.eachSupportedId())(
'selects the right rows %p',
async databaseId => {
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
await knex
.insert([
{
entity_id: '1',
entity_ref: 'k:ns/no_stitch_time',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
next_stitch_at: null,
next_stitch_ticket: null,
},
{
entity_id: '2',
entity_ref: 'k:ns/future_stitch_time',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
next_stitch_at: '2037-01-01T00:00:00.000',
next_stitch_ticket: 't1',
},
{
entity_id: '3',
entity_ref: 'k:ns/past_stitch_time',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
next_stitch_at: '1971-01-01T00:00:00.000',
next_stitch_ticket: 't3',
},
{
entity_id: '4',
entity_ref: 'k:ns/past_stitch_time_again',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
next_stitch_at: '1972-01-01T00:00:00.000',
next_stitch_ticket: 't4',
},
])
.into('refresh_state');
const rowsBefore = await knex('refresh_state');
const items = await getDeferredStitchableEntities({
knex,
batchSize: 1,
stitchTimeout: { seconds: 2 },
});
const rowsAfter = await knex('refresh_state');
expect(items).toEqual([
{
entityRef: 'k:ns/past_stitch_time',
stitchTicket: 't3',
stitchRequestedAt: expect.anything(),
},
]);
const hitRowBefore = rowsBefore.filter(r => r.entity_id === '3')[0]
.next_stitch_at;
const hitRowAfter = rowsAfter.filter(r => r.entity_id === '3')[0]
.next_stitch_at;
const missRowBefore = rowsBefore.filter(r => r.entity_id === '4')[0]
.next_stitch_at;
const missRowAfter = rowsAfter.filter(r => r.entity_id === '4')[0]
.next_stitch_at;
expect(+new Date(hitRowAfter)).toBeGreaterThan(+new Date(hitRowBefore));
expect(+new Date(missRowAfter)).toEqual(+new Date(missRowBefore));
},
);
});
plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.ts
+105
View File
@@ -0,0 +1,105 @@
/*
* Copyright 2023 The Backstage Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import { durationToMilliseconds, HumanDuration } from '@backstage/types';
import { Knex } from 'knex';
import { DateTime } from 'luxon';
import { timestampToDateTime } from '../../conversion';
import { DbRefreshStateRow } from '../../tables';
// TODO(freben): There is no retry counter or similar. If items start
// perpetually crashing during stitching, they'll just get silently retried over
// and over again, for better or worse. This will be visible in metrics though.
/**
* Finds entities that are marked for deferred stitching.
*
* @remarks
*
* This assumes that the stitching strategy is set to deferred.
*
* They are expected to already have the next_stitch_ticket set (by
* markForStitching) so that their tickets can be returned with each item.
*
* All returned items have their next_stitch_at updated to be moved forward by
* the given timeout duration. This has the effect that they will be picked up
* for stitching again in the future, if it hasn't completed by that point for
* some reason (restarts, crashes, etc).
*/
export async function getDeferredStitchableEntities(options: {
knex: Knex | Knex.Transaction;
batchSize: number;
stitchTimeout: HumanDuration;
}): Promise<
Array<{
entityRef: string;
stitchTicket: string;
stitchRequestedAt: DateTime; // the time BEFORE moving it forward by the timeout
}>
> {
const { knex, batchSize, stitchTimeout } = options;
let itemsQuery = knex<DbRefreshStateRow>('refresh_state').select(
'entity_ref',
'next_stitch_at',
'next_stitch_ticket',
);
// This avoids duplication of work because of race conditions and is
// also fast because locked rows are ignored rather than blocking.
// It's only available in MySQL and PostgreSQL
if (['mysql', 'mysql2', 'pg'].includes(knex.client.config.client)) {
itemsQuery = itemsQuery.forUpdate().skipLocked();
}
const items = await itemsQuery
.whereNotNull('next_stitch_at')
.whereNotNull('next_stitch_ticket')
.where('next_stitch_at', '<=', knex.fn.now())
.orderBy('next_stitch_at', 'asc')
.limit(batchSize);
if (!items.length) {
return [];
}
await knex<DbRefreshStateRow>('refresh_state')
.whereIn(
'entity_ref',
items.map(i => i.entity_ref),
)
// avoid race condition where someone completes a stitch right between these statements
.whereNotNull('next_stitch_ticket')
.update({
next_stitch_at: nowPlus(knex, stitchTimeout),
});
return items.map(i => ({
entityRef: i.entity_ref,
stitchTicket: i.next_stitch_ticket!,
stitchRequestedAt: timestampToDateTime(i.next_stitch_at!),
}));
}
function nowPlus(knex: Knex, duration: HumanDuration): Knex.Raw {
const seconds = durationToMilliseconds(duration) / 1000;
if (knex.client.config.client.includes('sqlite3')) {
return knex.raw(`datetime('now', ?)`, [`${seconds} seconds`]);
} else if (knex.client.config.client.includes('mysql')) {
return knex.raw(`now() + interval ${seconds} second`);
}
return knex.raw(`now() + interval '${seconds} seconds'`);
}
plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.test.ts
+75
View File
@@ -0,0 +1,75 @@
/*
* Copyright 2023 The Backstage Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import { TestDatabases } from '@backstage/backend-test-utils';
import { applyDatabaseMigrations } from '../../migrations';
import { markDeferredStitchCompleted } from './markDeferredStitchCompleted';
import { DbRefreshStateRow } from '../../tables';
jest.setTimeout(60_000);
describe('markDeferredStitchCompleted', () => {
const databases = TestDatabases.create({
ids: ['MYSQL_8', 'POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'],
});
it.each(databases.eachSupportedId())(
'completes only if unchanged %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/n',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
next_stitch_at: '1971-01-01T00:00:00.000',
next_stitch_ticket: 'the-ticket',
},
]);
async function result() {
return knex<DbRefreshStateRow>('refresh_state').select(
'next_stitch_at',
'next_stitch_ticket',
);
}
await markDeferredStitchCompleted({
knex,
entityRef: 'k:ns/n',
stitchTicket: 'the-wrong-ticket',
});
await expect(result()).resolves.toEqual([
{ next_stitch_at: expect.anything(), next_stitch_ticket: 'the-ticket' },
]);
await markDeferredStitchCompleted({
knex,
entityRef: 'k:ns/n',
stitchTicket: 'the-ticket',
});
await expect(result()).resolves.toEqual([
{ next_stitch_at: null, next_stitch_ticket: null },
]);
},
);
});
plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.ts
+46
View File
@@ -0,0 +1,46 @@
/*
* Copyright 2023 The Backstage Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import { Knex } from 'knex';
import { DbRefreshStateRow } from '../../tables';
/**
* Marks a single entity as having been stitched.
*
* @remarks
*
* This assumes that the stitching strategy is set to deferred.
*
* The timestamp and ticket are only reset if the ticket hasn't changed. If it
* has, it means that a new stitch request has been made, and the entity should
* be stitched once more some time in the future - or is indeed already being
* stitched concurrently with ourselves.
*/
export async function markDeferredStitchCompleted(option: {
knex: Knex | Knex.Transaction;
entityRef: string;
stitchTicket: string;
}): Promise<void> {
const { knex, entityRef, stitchTicket } = option;
await knex<DbRefreshStateRow>('refresh_state')
.update({
next_stitch_at: null,
next_stitch_ticket: null,
})
.where('entity_ref', '=', entityRef)
.andWhere('next_stitch_ticket', '=', stitchTicket);
}
plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts
+176
View File
@@ -26,6 +26,182 @@ describe('markForStitching', () => {
ids: ['MYSQL_8', 'POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'],
});
it.each(databases.eachSupportedId())(
'marks the right rows in deferred mode %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(),
next_stitch_at: null,
next_stitch_ticket: null,
},
{
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(),
next_stitch_at: null,
next_stitch_ticket: null,
},
{
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(),
next_stitch_at: null,
next_stitch_ticket: null,
},
{
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(),
next_stitch_at: '1971-01-01T00:00:00.000',
next_stitch_ticket: 'old',
},
]);
async function result() {
return knex<DbRefreshStateRow>('refresh_state')
.select('entity_id', 'next_stitch_at', 'next_stitch_ticket')
.orderBy('entity_id', 'asc');
}
const original = await result();
await markForStitching({
knex,
strategy: {
mode: 'deferred',
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
},
entityRefs: new Set(),
});
await expect(result()).resolves.toEqual([
{ entity_id: '1', next_stitch_at: null, next_stitch_ticket: null },
{ entity_id: '2', next_stitch_at: null, next_stitch_ticket: null },
{ entity_id: '3', next_stitch_at: null, next_stitch_ticket: null },
{
entity_id: '4',
next_stitch_at: expect.anything(),
next_stitch_ticket: 'old',
},
]);
await markForStitching({
knex,
strategy: {
mode: 'deferred',
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
},
entityRefs: new Set(['k:ns/one']),
});
await expect(result()).resolves.toEqual([
{
entity_id: '1',
next_stitch_at: expect.anything(),
next_stitch_ticket: expect.anything(),
},
{ entity_id: '2', next_stitch_at: null, next_stitch_ticket: null },
{ entity_id: '3', next_stitch_at: null, next_stitch_ticket: null },
{
entity_id: '4',
next_stitch_at: expect.anything(),
next_stitch_ticket: 'old',
},
]);
await markForStitching({
knex,
strategy: {
mode: 'deferred',
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
},
entityRefs: ['k:ns/two'],
});
await expect(result()).resolves.toEqual([
{
entity_id: '1',
next_stitch_at: expect.anything(),
next_stitch_ticket: expect.anything(),
},
{
entity_id: '2',
next_stitch_at: expect.anything(),
next_stitch_ticket: expect.anything(),
},
{ entity_id: '3', next_stitch_at: null, next_stitch_ticket: null },
{
entity_id: '4',
next_stitch_at: expect.anything(),
next_stitch_ticket: 'old',
},
]);
await markForStitching({
knex,
strategy: {
mode: 'deferred',
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
},
entityIds: ['3', '4'],
});
await expect(result()).resolves.toEqual([
{
entity_id: '1',
next_stitch_at: expect.anything(),
next_stitch_ticket: expect.anything(),
},
{
entity_id: '2',
next_stitch_at: expect.anything(),
next_stitch_ticket: expect.anything(),
},
{
entity_id: '3',
next_stitch_at: expect.anything(),
next_stitch_ticket: expect.anything(),
},
{
entity_id: '4',
next_stitch_at: expect.anything(),
next_stitch_ticket: expect.anything(),
},
]);
// It overwrites timers and tickets if they existed before
const final = await result();
for (let i = 0; i < final.length; ++i) {
expect(original[i].next_stitch_at).not.toEqual(final[i].next_stitch_at);
expect(original[i].next_stitch_ticket).not.toEqual(
final[i].next_stitch_ticket,
);
}
},
);
it.each(databases.eachSupportedId())(
'marks the right rows in immediate mode %p',
async databaseId => {
plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts
+63 -35
View File
@@ -16,11 +16,12 @@
import { Knex } from 'knex';
import splitToChunks from 'lodash/chunk';
import { v4 as uuid } from 'uuid';
import { StitchingStrategy } from '../../../stitching/types';
import { DbFinalEntitiesRow, DbRefreshStateRow } from '../../tables';
/**
* Marks a number of entities for deferred stitching some time in the near
* Marks a number of entities for stitching some time in the near
* future.
*
* @remarks
@@ -35,42 +36,69 @@ export async function markForStitching(options: {
const entityRefs = split(options.entityRefs);
const entityIds = split(options.entityIds);
const knex = options.knex;
const mode = options.strategy.mode;
for (const chunk of entityRefs) {
await knex
.table<DbFinalEntitiesRow>('final_entities')
.update({
hash: 'force-stitching',
})
.whereIn(
'entity_id',
knex<DbRefreshStateRow>('refresh_state')
.select('entity_id')
.whereIn('entity_ref', chunk),
);
await knex
.table<DbRefreshStateRow>('refresh_state')
.update({
result_hash: 'force-stitching',
next_update_at: knex.fn.now(),
})
.whereIn('entity_ref', chunk);
}
if (mode === 'immediate') {
for (const chunk of entityRefs) {
await knex
.table<DbFinalEntitiesRow>('final_entities')
.update({
hash: 'force-stitching',
})
.whereIn(
'entity_id',
knex<DbRefreshStateRow>('refresh_state')
.select('entity_id')
.whereIn('entity_ref', chunk),
);
await knex
.table<DbRefreshStateRow>('refresh_state')
.update({
result_hash: 'force-stitching',
next_update_at: knex.fn.now(),
})
.whereIn('entity_ref', chunk);
}
for (const chunk of entityIds) {
await knex
.table<DbFinalEntitiesRow>('final_entities')
.update({
hash: 'force-stitching',
})
.whereIn('entity_id', chunk);
await knex
.table<DbRefreshStateRow>('refresh_state')
.update({
result_hash: 'force-stitching',
next_update_at: knex.fn.now(),
})
.whereIn('entity_id', chunk);
for (const chunk of entityIds) {
await knex
.table<DbFinalEntitiesRow>('final_entities')
.update({
hash: 'force-stitching',
})
.whereIn('entity_id', chunk);
await knex
.table<DbRefreshStateRow>('refresh_state')
.update({
result_hash: 'force-stitching',
next_update_at: knex.fn.now(),
})
.whereIn('entity_id', chunk);
}
} else if (mode === 'deferred') {
// It's OK that this is shared across refresh state rows; it just needs to
// be uniquely generated for every new stitch request.
const ticket = uuid();
for (const chunk of entityRefs) {
await knex<DbRefreshStateRow>('refresh_state')
.update({
next_stitch_at: knex.fn.now(),
next_stitch_ticket: ticket,
})
.whereIn('entity_ref', chunk);
}
for (const chunk of entityIds) {
await knex<DbRefreshStateRow>('refresh_state')
.update({
next_stitch_at: knex.fn.now(),
next_stitch_ticket: ticket,
})
.whereIn('entity_id', chunk);
}
} else {
throw new Error(`Unknown stitching strategy mode ${mode}`);
}
}
plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts
+17 -4
View File
@@ -35,8 +35,9 @@ describe('performStitching', () => {
});
const logger = getVoidLogger();
// NOTE(freben): Testing the deferred path since it's a superset of the immediate one
it.each(databases.eachSupportedId())(
'runs the happy path in immediate mode for %p',
'runs the happy path in deferred mode for %p',
async databaseId => {
const knex = await databases.init(databaseId);
await applyDatabaseMigrations(knex);
@@ -87,7 +88,11 @@ describe('performStitching', () => {
await performStitching({
knex,
logger,
strategy: { mode: 'immediate' },
strategy: {
mode: 'deferred',
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
},
entityRef: 'k:ns/n',
});
@@ -172,7 +177,11 @@ describe('performStitching', () => {
await performStitching({
knex,
logger,
strategy: { mode: 'immediate' },
strategy: {
mode: 'deferred',
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
},
entityRef: 'k:ns/n',
});
@@ -195,7 +204,11 @@ describe('performStitching', () => {
await performStitching({
knex,
logger,
strategy: { mode: 'immediate' },
strategy: {
mode: 'deferred',
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 1 },
},
entityRef: 'k:ns/n',
});
plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts
+9
View File
@@ -32,6 +32,7 @@ import {
DbSearchRow,
} from '../../tables';
import { buildEntitySearch } from './buildEntitySearch';
import { markDeferredStitchCompleted } from './markDeferredStitchCompleted';
import { BATCH_SIZE, generateStableHash } from './util';
// See https://github.com/facebook/react/blob/f0cf832e1d0c8544c36aa8b310960885a11a847c/packages/react-dom-bindings/src/shared/sanitizeURL.js
@@ -217,6 +218,14 @@ export async function performStitching(options: {
.onConflict('entity_id')
.merge(['final_entity', 'hash', 'last_updated_at']);
if (options.strategy.mode === 'deferred') {
await markDeferredStitchCompleted({
knex: knex,
entityRef,
stitchTicket,
});
}
if (amountOfRowsChanged === 0) {
logger.debug(`Entity ${entityRef} is already stitched, skipping write.`);
return 'abandoned';
plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts
+110 -7
View File
@@ -101,7 +101,7 @@ describe('deleteOrphanedEntities', () => {
async function refreshState(knex: Knex) {
return await knex<DbRefreshStateRow>('refresh_state')
.orderBy('entity_ref')
.select('entity_ref', 'result_hash');
.select('entity_ref', 'result_hash', 'next_stitch_at');
}
async function finalEntities(knex: Knex) {
@@ -115,6 +115,7 @@ describe('deleteOrphanedEntities', () => {
.select({
entity_ref: 'refresh_state.entity_ref',
hash: 'final_entities.hash',
next_stitch_at: 'refresh_state.next_stitch_at',
});
}
@@ -178,30 +179,132 @@ describe('deleteOrphanedEntities', () => {
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: 'E1', result_hash: 'original', next_stitch_at: null },
{
entity_ref: 'E2',
result_hash: 'force-stitching',
next_stitch_at: null,
},
{
entity_ref: 'E7',
result_hash: 'force-stitching',
next_stitch_at: null,
},
{ entity_ref: 'E8', result_hash: 'original' },
{ entity_ref: 'E9', result_hash: 'original' },
{ entity_ref: 'E8', result_hash: 'original', next_stitch_at: null },
{ entity_ref: 'E9', result_hash: 'original', next_stitch_at: null },
]);
await expect(finalEntities(knex)).resolves.toEqual([
{ entity_ref: 'E1', hash: 'original' },
{ 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' },
{ entity_ref: 'E9', hash: 'original' },
{ entity_ref: 'E8', hash: 'original', next_stitch_at: null },
{ entity_ref: 'E9', hash: 'original', next_stitch_at: null },
]);
},
);
it.each(databases.eachSupportedId())(
'works for some mixed paths in deferred mode, %p',
async databaseId => {
/*
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(databaseId);
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(refreshState(knex)).resolves.toEqual([
{ entity_ref: 'E1', result_hash: 'original', next_stitch_at: null },
{
entity_ref: 'E2',
result_hash: 'original',
next_stitch_at: expect.anything(),
},
{
entity_ref: 'E7',
result_hash: 'original',
next_stitch_at: expect.anything(),
},
{ entity_ref: 'E8', result_hash: 'original', next_stitch_at: null },
{ entity_ref: 'E9', result_hash: 'original', next_stitch_at: null },
]);
await expect(finalEntities(knex)).resolves.toEqual([
{ entity_ref: 'E1', hash: 'original', next_stitch_at: null },
{
entity_ref: 'E2',
hash: 'original',
next_stitch_at: expect.anything(),
},
{
entity_ref: 'E7',
hash: 'original',
next_stitch_at: expect.anything(),
},
{ entity_ref: 'E8', hash: 'original', next_stitch_at: null },
{ entity_ref: 'E9', hash: 'original', next_stitch_at: null },
]);
},
);
plugins/catalog-backend/src/database/tables.ts
+41
View File
@@ -81,6 +81,47 @@ export type DbRefreshStateRow = {
* continuously gets moved forward as items are picked up for processing.
*/
next_update_at: string | Date;
/**
* If a stitch has been requested, this is the point in time that that last
* happened.
*
* @remarks
*
* Each time that a request is made, this timestamp is updated to the current
* time, overwriting the previous value if applicable.
*
* When the stitch loop runs and picks up an entity, this timestamp is not
* immediately reset. It's instead moved forward in time by a certain amount,
* which means that if the stitcher for some reason fails (eg if the process
* crashes or gets shut down), the entity will be picked up again in the
* future.
*
* Only when a stitch run is completed successfully, AND it's found that the
* stitch ticket has not changed since the start (which means that no new
* request has been made behind our backs), does the timestamp (and the
* ticket) get reset.
*/
next_stitch_at?: string | Date | null;
/**
* If a stitch has been requested, this is the unique ticket that was chosen
* to mark the last request.
*
* @remarks
*
* Each time that a request is made, a new random ticket is chosen,
* overwriting the previous value if applicable.
*
* When the stitch loop runs and picks up an entity, this column is left
* unchanged. This means that if the stitcher for some reason fails (eg if the
* process crashes or gets shut down), the entity will be picked up again in
* the future.
*
* Only when a stitch run is completed successfully, AND it's found that the
* stitch ticket has not changed since the start (which means that no new
* request has been made behind our backs), does the ticket (and the
* timestamp) get reset.
*/
next_stitch_ticket?: string | null;
/**
* The last time that this entity was emitted by somebody (the entity provider
* or a parent entity).
plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts
+4 -2
View File
@@ -28,7 +28,7 @@ import { metrics, trace } from '@opentelemetry/api';
import { ProcessingDatabase, RefreshStateItem } from '../database/types';
import { createCounterMetric, createSummaryMetric } from '../util/metrics';
import { CatalogProcessingOrchestrator, EntityProcessingResult } from './types';
import { Stitcher } from '../stitching/types';
import { Stitcher, stitchingStrategyFromConfig } from '../stitching/types';
import { startTaskPipeline } from './TaskPipeline';
import { PluginTaskScheduler } from '@backstage/backend-tasks';
import { Config } from '@backstage/config';
@@ -335,11 +335,13 @@ export class DefaultCatalogProcessingEngine {
return () => {};
}
const stitchingStrategy = stitchingStrategyFromConfig(this.config);
const runOnce = async () => {
try {
const n = await deleteOrphanedEntities({
knex: this.knex,
strategy: { mode: 'immediate' },
strategy: stitchingStrategy,
});
if (n > 0) {
this.logger.info(`Deleted ${n} orphaned entities`);
plugins/catalog-backend/src/stitching/DefaultStitcher.ts
+72 -5
View File
@@ -15,14 +15,26 @@
*/
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 { Logger } from 'winston';
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 } from './types';
import {
Stitcher,
StitchingStrategy,
stitchingStrategyFromConfig,
} from './types';
type DeferredStitchItem = Awaited<
ReturnType<typeof getDeferredStitchableEntities>
>[0];
type StitchProgressTracker = ReturnType<typeof progressTracker>;
@@ -36,9 +48,10 @@ export class DefaultStitcher implements Stitcher {
private readonly logger: Logger;
private readonly strategy: StitchingStrategy;
private readonly tracker: StitchProgressTracker;
private stopFunc?: () => void;
static fromConfig(
_config: Config,
config: Config,
options: {
knex: Knex;
logger: Logger;
@@ -47,7 +60,7 @@ export class DefaultStitcher implements Stitcher {
return new DefaultStitcher({
knex: options.knex,
logger: options.logger,
strategy: { mode: 'immediate' },
strategy: stitchingStrategyFromConfig(config),
});
}
@@ -68,6 +81,16 @@ export class DefaultStitcher implements Stitcher {
}) {
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 });
@@ -91,11 +114,55 @@ export class DefaultStitcher implements Stitcher {
}
async start() {
// Only called immediately for now
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();
};
}
}
async stop() {
// Only called immediately for now
if (this.strategy.mode === 'deferred') {
if (this.stopFunc) {
this.stopFunc();
this.stopFunc = undefined;
}
}
}
async #getStitchableEntities(count: number, stitchTimeout: HumanDuration) {
try {
return await getDeferredStitchableEntities({
knex: this.knex,
batchSize: count,
stitchTimeout: stitchTimeout,
});
} catch (error) {
this.logger.warn('Failed to load stitchable entities', error);
return [];
}
}
async #stitchOne(options: {
plugins/catalog-backend/src/stitching/progressTracker.ts
+27 -1
View File
@@ -19,10 +19,11 @@ import { metrics } from '@opentelemetry/api';
import { Knex } from 'knex';
import { DateTime } from 'luxon';
import { Logger } from 'winston';
import { DbRefreshStateRow } from '../database/tables';
import { createCounterMetric } from '../util/metrics';
// Helps wrap the timing and logging behaviors
export function progressTracker(_knex: Knex, logger: Logger) {
export function progressTracker(knex: Knex, logger: Logger) {
// prom-client metrics are deprecated in favour of OpenTelemetry metrics.
const promStitchedEntities = createCounterMetric({
name: 'catalog_stitched_entities_count',
@@ -46,6 +47,26 @@ export function progressTracker(_knex: Knex, logger: Logger) {
},
);
const stitchingQueueCount = meter.createObservableGauge(
'catalog.stitching.queue.length',
{ description: 'Number of entities currently in the stitching queue' },
);
stitchingQueueCount.addCallback(async result => {
const total = await knex<DbRefreshStateRow>('refresh_state')
.count({ count: '*' })
.whereNotNull('next_stitch_at');
result.observe(Number(total[0].count));
});
const stitchingQueueDelay = meter.createHistogram(
'catalog.stitching.queue.delay',
{
description:
'The amount of delay between being scheduled for stitching, and the start of actually being stitched',
unit: 'seconds',
},
);
function stitchStart(item: {
entityRef: string;
stitchRequestedAt?: DateTime;
@@ -53,6 +74,11 @@ export function progressTracker(_knex: Knex, logger: Logger) {
logger.debug(`Stitching ${item.entityRef}`);
const startTime = process.hrtime();
if (item.stitchRequestedAt) {
stitchingQueueDelay.record(
-item.stitchRequestedAt.diffNow().as('seconds'),
);
}
function endTime() {
const delta = process.hrtime(startTime);
plugins/catalog-backend/src/stitching/types.ts
+44 -4
View File
@@ -14,6 +14,9 @@
* limitations under the License.
*/
import { Config } 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
@@ -33,8 +36,45 @@ export interface Stitcher {
* @remarks
*
* In immediate mode, stitching happens "in-band" (blocking) immediately when
* each processing task finishes.
* 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.
*/
export type StitchingStrategy = {
mode: 'immediate';
};
export type StitchingStrategy =
| {
mode: 'immediate';
}
| {
mode: 'deferred';
pollingInterval: HumanDuration;
stitchTimeout: HumanDuration;
};
export function stitchingStrategyFromConfig(config: Config): StitchingStrategy {
const strategyMode = config.getOptionalString(
'catalog.stitchingStrategy.mode',
);
if (strategyMode === undefined || strategyMode === 'immediate') {
return {
mode: 'immediate',
};
} else if (strategyMode === 'deferred') {
// TODO(freben): Make parameters configurable
return {
mode: 'deferred',
pollingInterval: { seconds: 1 },
stitchTimeout: { seconds: 60 },
};
}
throw new Error(
`Invalid stitching strategy mode '${strategyMode}', expected one of 'immediate' or 'deferred'`,
);
}
plugins/catalog-backend/src/tests/integration.test.ts
+5
View File
@@ -215,6 +215,11 @@ class TestHarness {
connection: ':memory:',
},
},
catalog: {
stitchingStrategy: {
mode: 'immediate',
},
},
},
);
const logger = options?.logger ?? getVoidLogger();
plugins/catalog-backend/src/tests/migrations.test.ts
+71
View File
@@ -239,4 +239,75 @@ describe('migrations', () => {
await knex.destroy();
},
);
it.each(databases.eachSupportedId())(
'20230525141717_stitch_queue.js, %p',
async databaseId => {
const knex = await databases.init(databaseId);
await migrateUntilBefore(knex, '20230525141717_stitch_queue.js');
await knex
.insert([
{
entity_id: 'my-id',
entity_ref: 'k:ns/n',
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
next_update_at: knex.fn.now(),
last_discovery_at: knex.fn.now(),
},
])
.into('refresh_state');
await knex
.insert({
entity_id: 'my-id',
hash: 'd1c0c56d5fea4238e4c091d9dde4cb42d05a6b7e',
stitch_ticket: '',
final_entity: '{}',
})
.into('final_entities');
await migrateUpOnce(knex);
await expect(knex('refresh_state')).resolves.toEqual([
{
entity_id: 'my-id',
entity_ref: 'k:ns/n',
location_key: null,
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
cache: null,
unprocessed_hash: null,
result_hash: null,
next_update_at: expect.anything(),
next_stitch_at: null,
next_stitch_ticket: null,
last_discovery_at: expect.anything(),
},
]);
await migrateDownOnce(knex);
await expect(knex('refresh_state')).resolves.toEqual([
{
entity_id: 'my-id',
entity_ref: 'k:ns/n',
location_key: null,
unprocessed_entity: '{}',
processed_entity: '{}',
errors: '[]',
cache: null,
unprocessed_hash: null,
result_hash: null,
next_update_at: expect.anything(),
last_discovery_at: expect.anything(),
},
]);
await knex.destroy();
},
);
});
plugins/catalog-backend/src/tests/performance/stitchingPerformance.test.ts
+77 -15
View File
@@ -19,18 +19,24 @@ import {
createBackendModule,
createServiceFactory,
} from '@backstage/backend-plugin-api';
import { TestDatabases, startTestBackend } from '@backstage/backend-test-utils';
import {
mockServices,
startTestBackend,
TestDatabases,
} from '@backstage/backend-test-utils';
import { catalogProcessingExtensionPoint } from '@backstage/plugin-catalog-node/alpha';
import { Knex } from 'knex';
import { applyDatabaseMigrations } from '../../database/migrations';
import {
SyntheticLoadEntitiesProcessor,
SyntheticLoadEntitiesProvider,
SyntheticLoadEvents,
SyntheticLoadOptions,
SyntheticLoadEntitiesProvider,
SyntheticLoadEntitiesProcessor,
} from './lib/catalogModuleSyntheticLoadEntities';
import { describePerformanceTest, performanceTraceEnabled } from './lib/env';
jest.setTimeout(600_000);
function defer<T>() {
let resolve: (value: T | PromiseLike<T>) => void;
let reject: (error?: unknown) => void;
@@ -147,17 +153,6 @@ class Tracker {
}
}
function staticDatabase(knex: Knex) {
return createServiceFactory({
service: coreServices.database,
deps: {},
createRootContext: () => undefined,
factory: () => ({ getClient: async () => knex }),
});
}
jest.setTimeout(600_000);
describePerformanceTest('stitchingPerformance', () => {
const databases = TestDatabases.create({
ids: [/* 'MYSQL_8', */ 'POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'],
@@ -176,12 +171,79 @@ describePerformanceTest('stitchingPerformance', () => {
childrenCount: 3,
};
const config = {
backend: { baseUrl: 'http://localhost:7007' },
catalog: { stitchingStrategy: { mode: 'immediate' } },
};
const tracker = new Tracker(knex, load);
const backend = await startTestBackend({
features: [
import('@backstage/plugin-catalog-backend/alpha'),
staticDatabase(knex),
mockServices.rootConfig.factory({ data: config }),
createServiceFactory({
service: coreServices.database,
deps: {},
factory: () => ({ getClient: async () => knex }),
}),
createBackendModule({
moduleId: 'syntheticLoadEntities',
pluginId: 'catalog',
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.each(databases.eachSupportedId())(
'runs stitching in deferred mode, %p',
async databaseId => {
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: 'deferred' } },
};
const tracker = new Tracker(knex, load);
const backend = await startTestBackend({
features: [
import('@backstage/plugin-catalog-backend/alpha'),
mockServices.rootConfig.factory({ data: config }),
createServiceFactory({
service: coreServices.database,
deps: {},
factory: () => ({ getClient: async () => knex }),
}),
createBackendModule({
moduleId: 'syntheticLoadEntities',
pluginId: 'catalog',