From 8d756968f9c8676c391868a5a9aae931994c877b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Thu, 25 May 2023 14:40:12 +0200 Subject: [PATCH 1/2] implement the deferred stitching mode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- .changeset/loud-ghosts-deny.md | 16 ++ plugins/catalog-backend/config.d.ts | 13 ++ .../migrations/20230525141717_stitch_queue.js | 85 +++++++++ .../getDeferredStitchableEntities.test.ts | 114 ++++++++++++ .../stitcher/getDeferredStitchableEntities.ts | 105 +++++++++++ .../markDeferredStitchCompleted.test.ts | 75 ++++++++ .../stitcher/markDeferredStitchCompleted.ts | 46 +++++ .../stitcher/markForStitching.test.ts | 176 ++++++++++++++++++ .../operations/stitcher/markForStitching.ts | 95 ++++++---- .../stitcher/performStitching.test.ts | 21 ++- .../operations/stitcher/performStitching.ts | 9 + .../util/deleteOrphanedEntities.test.ts | 117 +++++++++++- .../catalog-backend/src/database/tables.ts | 41 ++++ .../DefaultCatalogProcessingEngine.ts | 6 +- .../src/stitching/DefaultStitcher.ts | 77 +++++++- .../src/stitching/progressTracker.ts | 28 ++- .../catalog-backend/src/stitching/types.ts | 48 ++++- .../src/tests/integration.test.ts | 5 + .../src/tests/migrations.test.ts | 71 +++++++ .../performance/stitchingPerformance.test.ts | 92 +++++++-- 20 files changed, 1167 insertions(+), 73 deletions(-) create mode 100644 .changeset/loud-ghosts-deny.md create mode 100644 plugins/catalog-backend/migrations/20230525141717_stitch_queue.js create mode 100644 plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.test.ts create mode 100644 plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.ts create mode 100644 plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.test.ts create mode 100644 plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.ts diff --git a/.changeset/loud-ghosts-deny.md b/.changeset/loud-ghosts-deny.md new file mode 100644 index 0000000000..e66b052c4e --- /dev/null +++ b/.changeset/loud-ghosts-deny.md @@ -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. diff --git a/plugins/catalog-backend/config.d.ts b/plugins/catalog-backend/config.d.ts index 4b9f606043..265d78b496 100644 --- a/plugins/catalog-backend/config.d.ts +++ b/plugins/catalog-backend/config.d.ts @@ -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. * diff --git a/plugins/catalog-backend/migrations/20230525141717_stitch_queue.js b/plugins/catalog-backend/migrations/20230525141717_stitch_queue.js new file mode 100644 index 0000000000..f4344a290d --- /dev/null +++ b/plugins/catalog-backend/migrations/20230525141717_stitch_queue.js @@ -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'); + }); +}; diff --git a/plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.test.ts new file mode 100644 index 0000000000..a909515f14 --- /dev/null +++ b/plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.test.ts @@ -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)); + }, + ); +}); diff --git a/plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.ts b/plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.ts new file mode 100644 index 0000000000..05c3f4d2eb --- /dev/null +++ b/plugins/catalog-backend/src/database/operations/stitcher/getDeferredStitchableEntities.ts @@ -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('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('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'`); +} diff --git a/plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.test.ts new file mode 100644 index 0000000000..76a6169cf4 --- /dev/null +++ b/plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.test.ts @@ -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('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('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 }, + ]); + }, + ); +}); diff --git a/plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.ts b/plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.ts new file mode 100644 index 0000000000..d1c7c3a6b8 --- /dev/null +++ b/plugins/catalog-backend/src/database/operations/stitcher/markDeferredStitchCompleted.ts @@ -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 { + const { knex, entityRef, stitchTicket } = option; + + await knex('refresh_state') + .update({ + next_stitch_at: null, + next_stitch_ticket: null, + }) + .where('entity_ref', '=', entityRef) + .andWhere('next_stitch_ticket', '=', stitchTicket); +} diff --git a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts index ce44fe15d8..2cae76040b 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts @@ -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('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('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 => { diff --git a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts index 7ab9bda1c0..9f4bcc49e7 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts @@ -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 @@ -36,41 +37,65 @@ export async function markForStitching(options: { const entityIds = split(options.entityIds); const knex = options.knex; - for (const chunk of entityRefs) { - await knex - .table('final_entities') - .update({ - hash: 'force-stitching', - }) - .whereIn( - 'entity_id', - knex('refresh_state') - .select('entity_id') - .whereIn('entity_ref', chunk), - ); - await knex - .table('refresh_state') - .update({ - result_hash: 'force-stitching', - next_update_at: knex.fn.now(), - }) - .whereIn('entity_ref', chunk); - } + if (options.strategy.mode === 'immediate') { + for (const chunk of entityRefs) { + await knex + .table('final_entities') + .update({ + hash: 'force-stitching', + }) + .whereIn( + 'entity_id', + knex('refresh_state') + .select('entity_id') + .whereIn('entity_ref', chunk), + ); + await knex + .table('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('final_entities') - .update({ - hash: 'force-stitching', - }) - .whereIn('entity_id', chunk); - await knex - .table('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('final_entities') + .update({ + hash: 'force-stitching', + }) + .whereIn('entity_id', chunk); + await knex + .table('refresh_state') + .update({ + result_hash: 'force-stitching', + next_update_at: knex.fn.now(), + }) + .whereIn('entity_id', chunk); + } + } else { + // 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('refresh_state') + .update({ + next_stitch_at: knex.fn.now(), + next_stitch_ticket: ticket, + }) + .whereIn('entity_ref', chunk); + } + + for (const chunk of entityIds) { + await knex('refresh_state') + .update({ + next_stitch_at: knex.fn.now(), + next_stitch_ticket: ticket, + }) + .whereIn('entity_id', chunk); + } } } diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts index a864cbdef4..a471f1dce2 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts @@ -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', }); diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts index bcb112690c..e4388813e8 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts @@ -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'; diff --git a/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts b/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts index 07ade1c4af..3dacc8f09b 100644 --- a/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts +++ b/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts @@ -101,7 +101,7 @@ describe('deleteOrphanedEntities', () => { async function refreshState(knex: Knex) { return await knex('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 }, ]); }, ); diff --git a/plugins/catalog-backend/src/database/tables.ts b/plugins/catalog-backend/src/database/tables.ts index a429dcde8d..57731e3d27 100644 --- a/plugins/catalog-backend/src/database/tables.ts +++ b/plugins/catalog-backend/src/database/tables.ts @@ -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). diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts index 72f92b9b40..3acf811559 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts @@ -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`); diff --git a/plugins/catalog-backend/src/stitching/DefaultStitcher.ts b/plugins/catalog-backend/src/stitching/DefaultStitcher.ts index 095d80ad83..990c5efd3b 100644 --- a/plugins/catalog-backend/src/stitching/DefaultStitcher.ts +++ b/plugins/catalog-backend/src/stitching/DefaultStitcher.ts @@ -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 +>[0]; type StitchProgressTracker = ReturnType; @@ -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({ + 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: { diff --git a/plugins/catalog-backend/src/stitching/progressTracker.ts b/plugins/catalog-backend/src/stitching/progressTracker.ts index e8e08dbc54..f82dbe378e 100644 --- a/plugins/catalog-backend/src/stitching/progressTracker.ts +++ b/plugins/catalog-backend/src/stitching/progressTracker.ts @@ -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('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); diff --git a/plugins/catalog-backend/src/stitching/types.ts b/plugins/catalog-backend/src/stitching/types.ts index 5f32a85b1f..9f7a5652dd 100644 --- a/plugins/catalog-backend/src/stitching/types.ts +++ b/plugins/catalog-backend/src/stitching/types.ts @@ -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'`, + ); +} diff --git a/plugins/catalog-backend/src/tests/integration.test.ts b/plugins/catalog-backend/src/tests/integration.test.ts index 8aeb95f49b..5511677851 100644 --- a/plugins/catalog-backend/src/tests/integration.test.ts +++ b/plugins/catalog-backend/src/tests/integration.test.ts @@ -215,6 +215,11 @@ class TestHarness { connection: ':memory:', }, }, + catalog: { + stitchingStrategy: { + mode: 'immediate', + }, + }, }, ); const logger = options?.logger ?? getVoidLogger(); diff --git a/plugins/catalog-backend/src/tests/migrations.test.ts b/plugins/catalog-backend/src/tests/migrations.test.ts index d2c00e31c0..aa305598aa 100644 --- a/plugins/catalog-backend/src/tests/migrations.test.ts +++ b/plugins/catalog-backend/src/tests/migrations.test.ts @@ -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(); + }, + ); }); diff --git a/plugins/catalog-backend/src/tests/performance/stitchingPerformance.test.ts b/plugins/catalog-backend/src/tests/performance/stitchingPerformance.test.ts index c1a869f9ad..ad659ce845 100644 --- a/plugins/catalog-backend/src/tests/performance/stitchingPerformance.test.ts +++ b/plugins/catalog-backend/src/tests/performance/stitchingPerformance.test.ts @@ -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() { let resolve: (value: T | PromiseLike) => 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', From 2bc82320e2cead47905f027b3fc3d1782d0958be Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Sat, 7 Oct 2023 17:13:55 +0200 Subject: [PATCH 2/2] strict mode check MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- .../src/database/operations/stitcher/markForStitching.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts index 9f4bcc49e7..ecc364a9cc 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts @@ -36,8 +36,9 @@ export async function markForStitching(options: { const entityRefs = split(options.entityRefs); const entityIds = split(options.entityIds); const knex = options.knex; + const mode = options.strategy.mode; - if (options.strategy.mode === 'immediate') { + if (mode === 'immediate') { for (const chunk of entityRefs) { await knex .table('final_entities') @@ -74,7 +75,7 @@ export async function markForStitching(options: { }) .whereIn('entity_id', chunk); } - } else { + } 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(); @@ -96,6 +97,8 @@ export async function markForStitching(options: { }) .whereIn('entity_id', chunk); } + } else { + throw new Error(`Unknown stitching strategy mode ${mode}`); } }