From b97e9790f0894f8f92c2541abd680c59fd42019e 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] Internal refactors as a path toward #18062 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- .changeset/polite-donuts-fail.md | 5 + .../src/database/DefaultProcessingDatabase.ts | 6 - .../stitcher}/buildEntitySearch.test.ts | 2 +- .../operations/stitcher}/buildEntitySearch.ts | 4 +- .../stitcher/markForStitching.test.ts | 260 ++++++++++++++++ .../operations/stitcher/markForStitching.ts | 82 +++++ .../stitcher/performStitching.test.ts | 287 ++++++++++++++++++ .../operations/stitcher/performStitching.ts | 236 ++++++++++++++ .../operations/stitcher}/util.ts | 0 .../util/deleteOrphanedEntities.test.ts | 36 ++- .../operations/util/deleteOrphanedEntities.ts | 34 +-- .../catalog-backend/src/database/tables.ts | 73 ++++- plugins/catalog-backend/src/database/types.ts | 2 - .../DefaultCatalogProcessingEngine.test.ts | 20 +- .../DefaultCatalogProcessingEngine.ts | 63 ++-- .../catalog-backend/src/processing/types.ts | 4 +- .../src/service/CatalogBuilder.ts | 20 +- .../service/DefaultEntitiesCatalog.test.ts | 10 +- .../src/service/DefaultEntitiesCatalog.ts | 7 +- .../src/service/DefaultRefreshService.test.ts | 9 +- ...itcher.test.ts => DefaultStitcher.test.ts} | 14 +- .../src/stitching/DefaultStitcher.ts | 124 ++++++++ .../catalog-backend/src/stitching/Stitcher.ts | 250 --------------- .../src/stitching/progressTracker.ts | 84 +++++ .../catalog-backend/src/stitching/types.ts | 40 +++ .../src/tests/integration.test.ts | 16 +- .../performance/stitchingPerformance.test.ts | 2 +- 27 files changed, 1335 insertions(+), 355 deletions(-) create mode 100644 .changeset/polite-donuts-fail.md rename plugins/catalog-backend/src/{stitching => database/operations/stitcher}/buildEntitySearch.test.ts (99%) rename plugins/catalog-backend/src/{stitching => database/operations/stitcher}/buildEntitySearch.ts (98%) create mode 100644 plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts create mode 100644 plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts create mode 100644 plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts create mode 100644 plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts rename plugins/catalog-backend/src/{stitching => database/operations/stitcher}/util.ts (100%) rename plugins/catalog-backend/src/stitching/{Stitcher.test.ts => DefaultStitcher.test.ts} (95%) create mode 100644 plugins/catalog-backend/src/stitching/DefaultStitcher.ts delete mode 100644 plugins/catalog-backend/src/stitching/Stitcher.ts create mode 100644 plugins/catalog-backend/src/stitching/progressTracker.ts create mode 100644 plugins/catalog-backend/src/stitching/types.ts diff --git a/.changeset/polite-donuts-fail.md b/.changeset/polite-donuts-fail.md new file mode 100644 index 0000000000..4b71dfddc3 --- /dev/null +++ b/.changeset/polite-donuts-fail.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend': patch +--- + +Internal refactors as a path toward #18062 diff --git a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts index 01025f73dd..2daf8e6fc5 100644 --- a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts @@ -42,7 +42,6 @@ import { import { checkLocationKeyConflict } from './operations/refreshState/checkLocationKeyConflict'; import { insertUnprocessedEntity } from './operations/refreshState/insertUnprocessedEntity'; import { updateUnprocessedEntity } from './operations/refreshState/updateUnprocessedEntity'; -import { deleteOrphanedEntities } from './operations/util/deleteOrphanedEntities'; import { generateStableHash } from './util'; import { EventBroker, EventParams } from '@backstage/plugin-events-node'; import { DateTime } from 'luxon'; @@ -275,11 +274,6 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { return { entityRefs }; } - async deleteOrphanedEntities(txOpaque: Transaction): Promise { - const tx = txOpaque as Knex.Transaction; - return await deleteOrphanedEntities({ tx }); - } - async transaction(fn: (tx: Transaction) => Promise): Promise { try { let result: T | undefined = undefined; diff --git a/plugins/catalog-backend/src/stitching/buildEntitySearch.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/buildEntitySearch.test.ts similarity index 99% rename from plugins/catalog-backend/src/stitching/buildEntitySearch.test.ts rename to plugins/catalog-backend/src/database/operations/stitcher/buildEntitySearch.test.ts index 384794d230..8bc5ba57a9 100644 --- a/plugins/catalog-backend/src/stitching/buildEntitySearch.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/buildEntitySearch.test.ts @@ -14,7 +14,7 @@ * limitations under the License. */ -import { Entity, DEFAULT_NAMESPACE } from '@backstage/catalog-model'; +import { DEFAULT_NAMESPACE, Entity } from '@backstage/catalog-model'; import { buildEntitySearch, mapToRows, traverse } from './buildEntitySearch'; describe('buildEntitySearch', () => { diff --git a/plugins/catalog-backend/src/stitching/buildEntitySearch.ts b/plugins/catalog-backend/src/database/operations/stitcher/buildEntitySearch.ts similarity index 98% rename from plugins/catalog-backend/src/stitching/buildEntitySearch.ts rename to plugins/catalog-backend/src/database/operations/stitcher/buildEntitySearch.ts index b97083f207..30f2987cb5 100644 --- a/plugins/catalog-backend/src/stitching/buildEntitySearch.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/buildEntitySearch.ts @@ -14,9 +14,9 @@ * limitations under the License. */ -import { Entity, DEFAULT_NAMESPACE } from '@backstage/catalog-model'; +import { DEFAULT_NAMESPACE, Entity } from '@backstage/catalog-model'; import { InputError } from '@backstage/errors'; -import { DbSearchRow } from '../database/tables'; +import { DbSearchRow } from '../../tables'; // These are excluded in the generic loop, either because they do not make sense // to index, or because they are special-case always inserted whether they are diff --git a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts new file mode 100644 index 0000000000..ce44fe15d8 --- /dev/null +++ b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.test.ts @@ -0,0 +1,260 @@ +/* + * 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 { markForStitching } from './markForStitching'; +import { DbFinalEntitiesRow, DbRefreshStateRow } from '../../tables'; + +jest.setTimeout(60_000); + +describe('markForStitching', () => { + const databases = TestDatabases.create({ + ids: ['MYSQL_8', 'POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'], + }); + + it.each(databases.eachSupportedId())( + 'marks the right rows in immediate 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: '{}', + result_hash: 'old', + errors: '[]', + next_update_at: knex.fn.now(), + last_discovery_at: knex.fn.now(), + }, + { + entity_id: '2', + entity_ref: 'k:ns/two', + unprocessed_entity: '{}', + processed_entity: '{}', + result_hash: 'old', + errors: '[]', + next_update_at: knex.fn.now(), + last_discovery_at: knex.fn.now(), + }, + { + entity_id: '3', + entity_ref: 'k:ns/three', + unprocessed_entity: '{}', + processed_entity: '{}', + result_hash: 'old', + errors: '[]', + next_update_at: knex.fn.now(), + last_discovery_at: knex.fn.now(), + }, + { + entity_id: '4', + entity_ref: 'k:ns/four', + unprocessed_entity: '{}', + processed_entity: '{}', + result_hash: 'old', + errors: '[]', + next_update_at: knex.fn.now(), + last_discovery_at: knex.fn.now(), + }, + ]); + await knex('final_entities').insert([ + { + entity_id: '1', + final_entity: '{}', + hash: 'old', + stitch_ticket: 'old', + }, + { + entity_id: '2', + final_entity: '{}', + hash: 'old', + stitch_ticket: 'old', + }, + { + entity_id: '3', + final_entity: '{}', + hash: 'old', + stitch_ticket: 'old', + }, + { + entity_id: '4', + final_entity: '{}', + hash: 'old', + stitch_ticket: 'old', + }, + ]); + + async function result() { + return knex('refresh_state') + .leftJoin( + 'final_entities', + 'final_entities.entity_id', + 'refresh_state.entity_id', + ) + .select({ + entity_id: 'refresh_state.entity_id', + next_update_at: 'refresh_state.next_update_at', + refresh_state_hash: 'refresh_state.result_hash', + final_entities_hash: 'final_entities.hash', + }) + .orderBy('entity_id', 'asc'); + } + + // Ensure that now() isn't evaluating to the same thing + await new Promise(resolve => setTimeout(resolve, 1100)); + + const original = await result(); + + await markForStitching({ + knex, + strategy: { mode: 'immediate' }, + entityRefs: new Set(), + }); + await expect(result()).resolves.toEqual([ + { + entity_id: '1', + next_update_at: expect.anything(), + refresh_state_hash: 'old', + final_entities_hash: 'old', + }, + { + entity_id: '2', + next_update_at: expect.anything(), + refresh_state_hash: 'old', + final_entities_hash: 'old', + }, + { + entity_id: '3', + next_update_at: expect.anything(), + refresh_state_hash: 'old', + final_entities_hash: 'old', + }, + { + entity_id: '4', + next_update_at: expect.anything(), + refresh_state_hash: 'old', + final_entities_hash: 'old', + }, + ]); + + await markForStitching({ + knex, + strategy: { mode: 'immediate' }, + entityRefs: new Set(['k:ns/one']), + }); + await expect(result()).resolves.toEqual([ + { + entity_id: '1', + next_update_at: expect.anything(), + refresh_state_hash: 'force-stitching', + final_entities_hash: 'force-stitching', + }, + { + entity_id: '2', + next_update_at: expect.anything(), + refresh_state_hash: 'old', + final_entities_hash: 'old', + }, + { + entity_id: '3', + next_update_at: expect.anything(), + refresh_state_hash: 'old', + final_entities_hash: 'old', + }, + { + entity_id: '4', + next_update_at: expect.anything(), + refresh_state_hash: 'old', + final_entities_hash: 'old', + }, + ]); + + await markForStitching({ + knex, + strategy: { mode: 'immediate' }, + entityRefs: ['k:ns/two'], + }); + await expect(result()).resolves.toEqual([ + { + entity_id: '1', + next_update_at: expect.anything(), + refresh_state_hash: 'force-stitching', + final_entities_hash: 'force-stitching', + }, + { + entity_id: '2', + next_update_at: expect.anything(), + refresh_state_hash: 'force-stitching', + final_entities_hash: 'force-stitching', + }, + { + entity_id: '3', + next_update_at: expect.anything(), + refresh_state_hash: 'old', + final_entities_hash: 'old', + }, + { + entity_id: '4', + next_update_at: expect.anything(), + refresh_state_hash: 'old', + final_entities_hash: 'old', + }, + ]); + + await markForStitching({ + knex, + strategy: { mode: 'immediate' }, + entityIds: ['3', '4'], + }); + await expect(result()).resolves.toEqual([ + { + entity_id: '1', + next_update_at: expect.anything(), + refresh_state_hash: 'force-stitching', + final_entities_hash: 'force-stitching', + }, + { + entity_id: '2', + next_update_at: expect.anything(), + refresh_state_hash: 'force-stitching', + final_entities_hash: 'force-stitching', + }, + { + entity_id: '3', + next_update_at: expect.anything(), + refresh_state_hash: 'force-stitching', + final_entities_hash: 'force-stitching', + }, + { + entity_id: '4', + next_update_at: expect.anything(), + refresh_state_hash: 'force-stitching', + final_entities_hash: 'force-stitching', + }, + ]); + + // It overwrites timers + const final = await result(); + for (let i = 0; i < final.length; ++i) { + expect(original[i].next_update_at).not.toEqual(final[i].next_update_at); + } + }, + ); +}); diff --git a/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts new file mode 100644 index 0000000000..7ab9bda1c0 --- /dev/null +++ b/plugins/catalog-backend/src/database/operations/stitcher/markForStitching.ts @@ -0,0 +1,82 @@ +/* + * 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 splitToChunks from 'lodash/chunk'; +import { StitchingStrategy } from '../../../stitching/types'; +import { DbFinalEntitiesRow, DbRefreshStateRow } from '../../tables'; + +/** + * Marks a number of entities for deferred stitching some time in the near + * future. + * + * @remarks + */ +export async function markForStitching(options: { + knex: Knex | Knex.Transaction; + strategy: StitchingStrategy; + entityRefs?: Iterable; + entityIds?: Iterable; +}): Promise { + // Splitting to chunks just to cover pathological cases that upset the db + const entityRefs = split(options.entityRefs); + 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); + } + + 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); + } +} + +function split(input: Iterable | undefined): string[][] { + if (!input) { + return []; + } + return splitToChunks(Array.isArray(input) ? input : [...input], 200); +} diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts new file mode 100644 index 0000000000..a864cbdef4 --- /dev/null +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.test.ts @@ -0,0 +1,287 @@ +/* + * Copyright 2021 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 { getVoidLogger } from '@backstage/backend-common'; +import { TestDatabases } from '@backstage/backend-test-utils'; +import { Entity } from '@backstage/catalog-model'; +import { applyDatabaseMigrations } from '../../migrations'; +import { + DbFinalEntitiesRow, + DbRefreshStateReferencesRow, + DbRefreshStateRow, + DbRelationsRow, + DbSearchRow, +} from '../../tables'; +import { performStitching } from './performStitching'; + +jest.setTimeout(60_000); + +describe('performStitching', () => { + const databases = TestDatabases.create({ + ids: ['MYSQL_8', 'POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'], + }); + const logger = getVoidLogger(); + + it.each(databases.eachSupportedId())( + 'runs the happy path in immediate mode for %p', + async databaseId => { + const knex = await databases.init(databaseId); + await applyDatabaseMigrations(knex); + + let entities: DbFinalEntitiesRow[]; + let entity: Entity; + + await knex('refresh_state').insert([ + { + entity_id: 'my-id', + entity_ref: 'k:ns/n', + unprocessed_entity: JSON.stringify({}), + processed_entity: JSON.stringify({ + apiVersion: 'a', + kind: 'k', + metadata: { + name: 'n', + namespace: 'ns', + }, + spec: { + k: 'v', + }, + }), + errors: '[]', + next_update_at: knex.fn.now(), + last_discovery_at: knex.fn.now(), + }, + ]); + await knex( + 'refresh_state_references', + ).insert([{ source_key: 'a', target_entity_ref: 'k:ns/n' }]); + await knex('relations').insert([ + { + originating_entity_id: 'my-id', + source_entity_ref: 'k:ns/n', + type: 'looksAt', + target_entity_ref: 'k:ns/other', + }, + // handles and ignores duplicates + { + originating_entity_id: 'my-id', + source_entity_ref: 'k:ns/n', + type: 'looksAt', + target_entity_ref: 'k:ns/other', + }, + ]); + + await performStitching({ + knex, + logger, + strategy: { mode: 'immediate' }, + entityRef: 'k:ns/n', + }); + + entities = await knex('final_entities'); + + expect(entities.length).toBe(1); + entity = JSON.parse(entities[0].final_entity!); + expect(entity).toEqual({ + relations: [ + { + type: 'looksAt', + targetRef: 'k:ns/other', + }, + ], + apiVersion: 'a', + kind: 'k', + metadata: { + name: 'n', + namespace: 'ns', + etag: expect.any(String), + uid: 'my-id', + }, + spec: { + k: 'v', + }, + }); + + expect(entity.metadata.etag).toEqual(entities[0].hash); + const last_updated_at = entities[0].last_updated_at; + expect(last_updated_at).not.toBeNull(); + const firstHash = entities[0].hash; + + const search = await knex('search'); + expect(search).toEqual( + expect.arrayContaining([ + { + entity_id: 'my-id', + key: 'relations.looksat', + original_value: 'k:ns/other', + value: 'k:ns/other', + }, + { + entity_id: 'my-id', + key: 'apiversion', + original_value: 'a', + value: 'a', + }, + { + entity_id: 'my-id', + key: 'kind', + original_value: 'k', + value: 'k', + }, + { + entity_id: 'my-id', + key: 'metadata.name', + original_value: 'n', + value: 'n', + }, + { + entity_id: 'my-id', + key: 'metadata.namespace', + original_value: 'ns', + value: 'ns', + }, + { + entity_id: 'my-id', + key: 'metadata.uid', + original_value: 'my-id', + value: 'my-id', + }, + { + entity_id: 'my-id', + key: 'spec.k', + original_value: 'v', + value: 'v', + }, + ]), + ); + + // Re-stitch without any changes + await performStitching({ + knex, + logger, + strategy: { mode: 'immediate' }, + entityRef: 'k:ns/n', + }); + + entities = await knex('final_entities'); + expect(entities.length).toBe(1); + entity = JSON.parse(entities[0].final_entity!); + expect(entities[0].hash).toEqual(firstHash); + expect(entity.metadata.etag).toEqual(firstHash); + + // Now add one more relation and re-stitch + await knex('relations').insert([ + { + originating_entity_id: 'my-id', + source_entity_ref: 'k:ns/n', + type: 'looksAt', + target_entity_ref: 'k:ns/third', + }, + ]); + + await performStitching({ + knex, + logger, + strategy: { mode: 'immediate' }, + entityRef: 'k:ns/n', + }); + + entities = await knex('final_entities'); + + expect(entities.length).toBe(1); + entity = JSON.parse(entities[0].final_entity!); + expect(entity).toEqual({ + relations: expect.arrayContaining([ + { + type: 'looksAt', + targetRef: 'k:ns/other', + }, + { + type: 'looksAt', + targetRef: 'k:ns/third', + }, + ]), + apiVersion: 'a', + kind: 'k', + metadata: { + name: 'n', + namespace: 'ns', + etag: expect.any(String), + uid: 'my-id', + }, + spec: { + k: 'v', + }, + }); + + expect(entities[0].hash).not.toEqual(firstHash); + expect(entities[0].hash).toEqual(entity.metadata.etag); + + expect(await knex('search')).toEqual( + expect.arrayContaining([ + { + entity_id: 'my-id', + key: 'relations.looksat', + original_value: 'k:ns/other', + value: 'k:ns/other', + }, + { + entity_id: 'my-id', + key: 'relations.looksat', + original_value: 'k:ns/third', + value: 'k:ns/third', + }, + { + entity_id: 'my-id', + key: 'apiversion', + original_value: 'a', + value: 'a', + }, + { + entity_id: 'my-id', + key: 'kind', + original_value: 'k', + value: 'k', + }, + { + entity_id: 'my-id', + key: 'metadata.name', + original_value: 'n', + value: 'n', + }, + { + entity_id: 'my-id', + key: 'metadata.namespace', + original_value: 'ns', + value: 'ns', + }, + { + entity_id: 'my-id', + key: 'metadata.uid', + original_value: 'my-id', + value: 'my-id', + }, + { + entity_id: 'my-id', + key: 'spec.k', + original_value: 'v', + value: 'v', + }, + ]), + ); + }, + ); +}); diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts new file mode 100644 index 0000000000..bcb112690c --- /dev/null +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts @@ -0,0 +1,236 @@ +/* + * 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 { ENTITY_STATUS_CATALOG_PROCESSING_TYPE } from '@backstage/catalog-client'; +import { + ANNOTATION_EDIT_URL, + ANNOTATION_VIEW_URL, + EntityRelation, +} from '@backstage/catalog-model'; +import { AlphaEntity, EntityStatusItem } from '@backstage/catalog-model/alpha'; +import { SerializedError } from '@backstage/errors'; +import { Knex } from 'knex'; +import { v4 as uuid } from 'uuid'; +import { Logger } from 'winston'; +import { StitchingStrategy } from '../../../stitching/types'; +import { + DbFinalEntitiesRow, + DbRefreshStateRow, + DbSearchRow, +} from '../../tables'; +import { buildEntitySearch } from './buildEntitySearch'; +import { BATCH_SIZE, generateStableHash } from './util'; + +// See https://github.com/facebook/react/blob/f0cf832e1d0c8544c36aa8b310960885a11a847c/packages/react-dom-bindings/src/shared/sanitizeURL.js +const scriptProtocolPattern = + // eslint-disable-next-line no-control-regex + /^[\u0000-\u001F ]*j[\r\n\t]*a[\r\n\t]*v[\r\n\t]*a[\r\n\t]*s[\r\n\t]*c[\r\n\t]*r[\r\n\t]*i[\r\n\t]*p[\r\n\t]*t[\r\n\t]*\:/i; + +/** + * Performs the act of stitching - to take all of the various outputs from the + * ingestion process, and stitching them together into the final entity JSON + * shape. + */ +export async function performStitching(options: { + knex: Knex | Knex.Transaction; + logger: Logger; + strategy: StitchingStrategy; + entityRef: string; + stitchTicket?: string; +}): Promise<'changed' | 'unchanged' | 'abandoned'> { + const { knex, logger, entityRef } = options; + const stitchTicket = options.stitchTicket ?? uuid(); + + const entityResult = await knex('refresh_state') + .where({ entity_ref: entityRef }) + .limit(1) + .select('entity_id'); + if (!entityResult.length) { + // Entity does no exist in refresh state table, no stitching required. + return 'abandoned'; + } + + // Insert stitching ticket that will be compared before inserting the final entity. + await knex('final_entities') + .insert({ + entity_id: entityResult[0].entity_id, + hash: '', + stitch_ticket: stitchTicket, + }) + .onConflict('entity_id') + .merge(['stitch_ticket']); + + // Selecting from refresh_state and final_entities should yield exactly + // one row (except in abnormal cases where the stitch was invoked for + // something that didn't exist at all, in which case it's zero rows). + // The join with the temporary incoming_references still gives one row. + const [processedResult, relationsResult] = await Promise.all([ + knex + .with('incoming_references', function incomingReferences(builder) { + return builder + .from('refresh_state_references') + .where({ target_entity_ref: entityRef }) + .count({ count: '*' }); + }) + .select({ + entityId: 'refresh_state.entity_id', + processedEntity: 'refresh_state.processed_entity', + errors: 'refresh_state.errors', + incomingReferenceCount: 'incoming_references.count', + previousHash: 'final_entities.hash', + }) + .from('refresh_state') + .where({ 'refresh_state.entity_ref': entityRef }) + .crossJoin(knex.raw('incoming_references')) + .leftOuterJoin('final_entities', { + 'final_entities.entity_id': 'refresh_state.entity_id', + }), + knex + .distinct({ + relationType: 'type', + relationTarget: 'target_entity_ref', + }) + .from('relations') + .where({ source_entity_ref: entityRef }) + .orderBy('relationType', 'asc') + .orderBy('relationTarget', 'asc'), + ]); + + // If there were no rows returned, it would mean that there was no + // matching row even in the refresh_state. This can happen for example + // if we emit a relation to something that hasn't been ingested yet. + // It's safe to ignore this stitch attempt in that case. + if (!processedResult.length) { + logger.debug( + `Unable to stitch ${entityRef}, item does not exist in refresh state table`, + ); + return 'abandoned'; + } + + const { + entityId, + processedEntity, + errors, + incomingReferenceCount, + previousHash, + } = processedResult[0]; + + // If there was no processed entity in place, the target hasn't been + // through the processing steps yet. It's safe to ignore this stitch + // attempt in that case, since another stitch will be triggered when + // that processing has finished. + if (!processedEntity) { + logger.debug( + `Unable to stitch ${entityRef}, the entity has not yet been processed`, + ); + return 'abandoned'; + } + + // Grab the processed entity and stitch all of the relevant data into + // it + const entity = JSON.parse(processedEntity) as AlphaEntity; + const isOrphan = Number(incomingReferenceCount) === 0; + let statusItems: EntityStatusItem[] = []; + + if (isOrphan) { + logger.debug(`${entityRef} is an orphan`); + entity.metadata.annotations = { + ...entity.metadata.annotations, + ['backstage.io/orphan']: 'true', + }; + } + if (errors) { + const parsedErrors = JSON.parse(errors) as SerializedError[]; + if (Array.isArray(parsedErrors) && parsedErrors.length) { + statusItems = parsedErrors.map(e => ({ + type: ENTITY_STATUS_CATALOG_PROCESSING_TYPE, + level: 'error', + message: `${e.name}: ${e.message}`, + error: e, + })); + } + } + // We opt to do this check here as we otherwise can't guarantee that it will be run after all processors + for (const annotation of [ANNOTATION_VIEW_URL, ANNOTATION_EDIT_URL]) { + const value = entity.metadata.annotations?.[annotation]; + if (typeof value === 'string' && scriptProtocolPattern.test(value)) { + entity.metadata.annotations![annotation] = + 'https://backstage.io/annotation-rejected-for-security-reasons'; + } + } + + // TODO: entityRef is lower case and should be uppercase in the final + // result + entity.relations = relationsResult + .filter(row => row.relationType /* exclude null row, if relevant */) + .map(row => ({ + type: row.relationType!, + targetRef: row.relationTarget!, + })); + if (statusItems.length) { + entity.status = { + ...entity.status, + items: [...(entity.status?.items ?? []), ...statusItems], + }; + } + + // If the output entity was actually not changed, just abort + const hash = generateStableHash(entity); + if (hash === previousHash) { + logger.debug(`Skipped stitching of ${entityRef}, no changes`); + return 'unchanged'; + } + + entity.metadata.uid = entityId; + if (!entity.metadata.etag) { + // If the original data source did not have its own etag handling, + // use the hash as a good-quality etag + entity.metadata.etag = hash; + } + + // This may throw if the entity is invalid, so we call it before + // the final_entities write, even though we may end up not needing + // to write the search index. + const searchEntries = buildEntitySearch(entityId, entity); + + const amountOfRowsChanged = await knex('final_entities') + .update({ + final_entity: JSON.stringify(entity), + hash, + last_updated_at: knex.fn.now(), + }) + .where('entity_id', entityId) + .where('stitch_ticket', stitchTicket) + .onConflict('entity_id') + .merge(['final_entity', 'hash', 'last_updated_at']); + + if (amountOfRowsChanged === 0) { + logger.debug(`Entity ${entityRef} is already stitched, skipping write.`); + return 'abandoned'; + } + + // TODO(freben): Search will probably need a similar safeguard against + // race conditions like the final_entities ticket handling above. + // Otherwise, it can be the case that: + // A writes the entity -> + // B writes the entity -> + // B writes search -> + // A writes search + await knex('search').where({ entity_id: entityId }).delete(); + await knex.batchInsert('search', searchEntries, BATCH_SIZE); + + return 'changed'; +} diff --git a/plugins/catalog-backend/src/stitching/util.ts b/plugins/catalog-backend/src/database/operations/stitcher/util.ts similarity index 100% rename from plugins/catalog-backend/src/stitching/util.ts rename to plugins/catalog-backend/src/database/operations/stitcher/util.ts 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 243b5db1b9..07ade1c4af 100644 --- a/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts +++ b/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts @@ -16,7 +16,7 @@ import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils'; import { Knex } from 'knex'; -import * as uuid from 'uuid'; +import { StitchingStrategy } from '../../../stitching/types'; import { applyDatabaseMigrations } from '../../migrations'; import { DbFinalEntitiesRow, @@ -39,14 +39,14 @@ describe('deleteOrphanedEntities', () => { return knex; } - async function run(knex: Knex): Promise { + async function run(knex: Knex, strategy: StitchingStrategy): Promise { let result: number; await knex.transaction( async tx => { // We can't return here, as knex swallows the return type in case the // transaction is rolled back: // https://github.com/knex/knex/blob/e37aeaa31c8ef9c1b07d2e4d3ec6607e557d800d/lib/transaction.js#L136 - result = await deleteOrphanedEntities({ tx }); + result = await deleteOrphanedEntities({ knex: tx, strategy }); }, { // If we explicitly trigger a rollback, don't fail. @@ -58,9 +58,8 @@ describe('deleteOrphanedEntities', () => { async function insertEntity(knex: Knex, ...entityRefs: string[]) { for (const ref of entityRefs) { - const entityId = uuid.v4(); await knex('refresh_state').insert({ - entity_id: entityId, + entity_id: `id-${ref}`, entity_ref: ref, unprocessed_entity: '{}', processed_entity: '{}', @@ -70,7 +69,7 @@ describe('deleteOrphanedEntities', () => { result_hash: 'original', }); await knex('final_entities').insert({ - entity_id: entityId, + entity_id: `id-${ref}`, hash: 'original', stitch_ticket: '', }); @@ -120,7 +119,7 @@ describe('deleteOrphanedEntities', () => { } it.each(databases.eachSupportedId())( - 'works for some mixed paths, %p', + 'works for some mixed paths in immediate mode, %p', async databaseId => { /* In this graph, edges represent refresh state references, not entity relations: @@ -176,18 +175,31 @@ describe('deleteOrphanedEntities', () => { await insertRelation(knex, 'E1', 'E2'); await insertRelation(knex, 'E2', 'E3'); await insertRelation(knex, 'E10', 'E6'); - await expect(run(knex)).resolves.toEqual(5); + await insertRelation(knex, 'E7', 'E6'); + await expect(run(knex, { mode: 'immediate' })).resolves.toEqual(5); await expect(refreshState(knex)).resolves.toEqual([ { entity_ref: 'E1', result_hash: 'original' }, - { entity_ref: 'E2', result_hash: 'orphan-relation-deleted' }, - { entity_ref: 'E7', result_hash: 'original' }, + { + entity_ref: 'E2', + result_hash: 'force-stitching', + }, + { + entity_ref: 'E7', + result_hash: 'force-stitching', + }, { entity_ref: 'E8', result_hash: 'original' }, { entity_ref: 'E9', result_hash: 'original' }, ]); await expect(finalEntities(knex)).resolves.toEqual([ { entity_ref: 'E1', hash: 'original' }, - { entity_ref: 'E2', hash: 'orphan-relation-deleted' }, - { entity_ref: 'E7', hash: 'original' }, + { + entity_ref: 'E2', + hash: 'force-stitching', + }, + { + entity_ref: 'E7', + hash: 'force-stitching', + }, { entity_ref: 'E8', hash: 'original' }, { entity_ref: 'E9', hash: 'original' }, ]); diff --git a/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.ts b/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.ts index eeab85f59e..5e2e46cf89 100644 --- a/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.ts +++ b/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.ts @@ -16,7 +16,9 @@ import { Knex } from 'knex'; import uniq from 'lodash/uniq'; -import { DbFinalEntitiesRow, DbRefreshStateRow } from '../../tables'; +import { StitchingStrategy } from '../../../stitching/types'; +import { DbRefreshStateRow } from '../../tables'; +import { markForStitching } from '../stitcher/markForStitching'; /** * Finds and deletes all orphaned entities, i.e. entities that do not have any @@ -24,15 +26,16 @@ import { DbFinalEntitiesRow, DbRefreshStateRow } from '../../tables'; * that would otherwise become orphaned. */ export async function deleteOrphanedEntities(options: { - tx: Knex.Transaction | Knex; + knex: Knex.Transaction | Knex; + strategy: StitchingStrategy; }): Promise { - const { tx } = options; + const { knex, strategy } = options; let total = 0; // Limit iterations for sanity for (let i = 0; i < 100; ++i) { - const candidates = await tx + const candidates = await knex .with('orphans', ['entity_id', 'entity_ref'], orphans => orphans .from('refresh_state') @@ -72,26 +75,17 @@ export async function deleteOrphanedEntities(options: { total += orphanIds.length; // Delete the orphans themselves - await tx + await knex .table('refresh_state') .delete() .whereIn('entity_id', orphanIds); - // Mark all of things that the orphans had relations to for processing and - // stitching - await tx - .table('final_entities') - .update({ - hash: 'orphan-relation-deleted', - }) - .whereIn('entity_id', orphanRelationIds); - await tx - .table('refresh_state') - .update({ - result_hash: 'orphan-relation-deleted', - next_update_at: tx.fn.now(), - }) - .whereIn('entity_id', orphanRelationIds); + // Mark all of the things that the orphans had relations to for stitching + await markForStitching({ + knex, + strategy, + entityIds: orphanRelationIds, + }); } return total; diff --git a/plugins/catalog-backend/src/database/tables.ts b/plugins/catalog-backend/src/database/tables.ts index e3f3efd1bb..a429dcde8d 100644 --- a/plugins/catalog-backend/src/database/tables.ts +++ b/plugins/catalog-backend/src/database/tables.ts @@ -29,17 +29,88 @@ export type DbLocationsRow = { target: string; }; +/** + * Represents the refresh_state table. + * + * @remarks + * + * Every unique entity ref emitted by a provider or a parent entity becomes a + * row in this table, even before processing has started on it. The actual final + * data, after processing and stitching completes, is instead in the + * final_entities table. + * + * Datetime columns are both string and Date, because different database engines + * return them in different forms on the client side. + */ export type DbRefreshStateRow = { + /** + * The unique ID of the entity. This is different to the entity ref, in that + * it gets regenerated randomly each time a row is added to the table, no + * matter what the original entity data was. + */ entity_id: string; + /** + * The entity string ref (on lowercase kind:namespace/name form) + */ entity_ref: string; + /** + * The JSON of the raw entity, as it was received from the entity provider. + */ unprocessed_entity: string; + /** + * A stable hash of the unprocessed entity, used to detect changed/unchanged + * data for a given entity over time. + */ unprocessed_hash?: string; + /** + * The JSON of the processed entity (if processing has run yet on it). + */ processed_entity?: string; + /** + * A stable hash of the processed entity AND all other emitted things during + * processing, such as relations. + */ result_hash?: string; + /** + * Per-entity cached data on JSON form. This is read and written by processors + * who wish to leverage this feature. + */ cache?: string; + /** + * The next point in time that this entity is due for processing. This + * continuously gets moved forward as items are picked up for processing. + */ next_update_at: string | Date; - last_discovery_at: string | Date; // remove? + /** + * The last time that this entity was emitted by somebody (the entity provider + * or a parent entity). + * + * @remarks + * + * Don't rely on this column more than at most as being loosely informative. + * Its semantics aren't fully settled yet. + */ + last_discovery_at: string | Date; + /** + * A JSON serialized array of errors (if any) encountered during processing. + */ errors?: string; + /** + * A conflict detection/resolution key for the entity. + * + * @remarks + * + * The exact value semantics differs, but may for example be a URL pointing to + * where the entity was sourced from. If a "competing" provider or parent + * entity tries to emit an entity that has the same entity ref but a different + * location key, a conflict is detected (you aren't allowed to "trample" over + * a previously existing entity). + * + * Some providers may choose to emit entities with no location key set at all. + * This is a signal that it's only loosely claimed, and that any other + * competing provider/parent is allowed to overwrite and claim it as theirs + * instead. + */ location_key?: string; }; diff --git a/plugins/catalog-backend/src/database/types.ts b/plugins/catalog-backend/src/database/types.ts index 592d5fe063..56e5f48f14 100644 --- a/plugins/catalog-backend/src/database/types.ts +++ b/plugins/catalog-backend/src/database/types.ts @@ -151,8 +151,6 @@ export interface ProcessingDatabase { txOpaque: Transaction, options: ListParentsOptions, ): Promise; - - deleteOrphanedEntities(txOpaque: Transaction): Promise; } /** diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts index 00235ef601..fba5e387a9 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts @@ -21,7 +21,7 @@ import waitForExpect from 'wait-for-expect'; import { DefaultProcessingDatabase } from '../database/DefaultProcessingDatabase'; import { DefaultCatalogProcessingEngine } from './DefaultCatalogProcessingEngine'; import { CatalogProcessingOrchestrator } from './types'; -import { Stitcher } from '../stitching/Stitcher'; +import { Stitcher } from '../stitching/types'; import { ConfigReader } from '@backstage/config'; describe('DefaultCatalogProcessingEngine', () => { @@ -66,6 +66,7 @@ describe('DefaultCatalogProcessingEngine', () => { config: new ConfigReader({}), logger: getVoidLogger(), processingDatabase: db, + knex: {} as any, orchestrator: orchestrator, stitcher: stitcher, createHash: () => hash, @@ -133,6 +134,7 @@ describe('DefaultCatalogProcessingEngine', () => { config: new ConfigReader({}), logger: getVoidLogger(), processingDatabase: db, + knex: {} as any, orchestrator: orchestrator, stitcher: stitcher, createHash: () => hash, @@ -216,6 +218,7 @@ describe('DefaultCatalogProcessingEngine', () => { config: new ConfigReader({}), logger: getVoidLogger(), processingDatabase: db, + knex: {} as any, orchestrator: orchestrator, stitcher: stitcher, createHash: () => hash, @@ -292,6 +295,7 @@ describe('DefaultCatalogProcessingEngine', () => { config: new ConfigReader({}), logger: getVoidLogger(), processingDatabase: db, + knex: {} as any, orchestrator: orchestrator, stitcher: stitcher, createHash: () => hash, @@ -350,6 +354,7 @@ describe('DefaultCatalogProcessingEngine', () => { config: new ConfigReader({}), logger: getVoidLogger(), processingDatabase: db, + knex: {} as any, orchestrator: orchestrator, stitcher: stitcher, createHash: () => hash, @@ -450,10 +455,10 @@ describe('DefaultCatalogProcessingEngine', () => { await waitForExpect(() => { expect(stitcher.stitch).toHaveBeenCalledTimes(2); }); - expect([...stitcher.stitch.mock.calls[0][0]]).toEqual( + expect([...stitcher.stitch.mock.calls[0][0].entityRefs!]).toEqual( expect.arrayContaining(['k:ns/me', 'k:ns/other1', 'k:ns/other2']), ); - expect([...stitcher.stitch.mock.calls[1][0]]).toEqual( + expect([...stitcher.stitch.mock.calls[1][0].entityRefs!]).toEqual( expect.arrayContaining(['k:ns/me', 'k:ns/other1', 'k:ns/other3']), ); await engine.stop(); @@ -464,6 +469,7 @@ describe('DefaultCatalogProcessingEngine', () => { config: new ConfigReader({}), logger: getVoidLogger(), processingDatabase: db, + knex: {} as any, orchestrator: orchestrator, stitcher: stitcher, createHash: () => hash, @@ -535,7 +541,7 @@ describe('DefaultCatalogProcessingEngine', () => { await waitForExpect(() => { expect(stitcher.stitch).toHaveBeenCalledTimes(1); }); - expect([...stitcher.stitch.mock.calls[0][0]]).toEqual( + expect([...stitcher.stitch.mock.calls[0][0].entityRefs!]).toEqual( expect.arrayContaining(['k:ns/me']), ); await engine.stop(); @@ -546,6 +552,7 @@ describe('DefaultCatalogProcessingEngine', () => { config: new ConfigReader({}), logger: getVoidLogger(), processingDatabase: db, + knex: {} as any, orchestrator: orchestrator, stitcher: stitcher, createHash: () => hash, @@ -622,7 +629,7 @@ describe('DefaultCatalogProcessingEngine', () => { await waitForExpect(() => { expect(stitcher.stitch).toHaveBeenCalledTimes(1); }); - expect([...stitcher.stitch.mock.calls[0][0]]).toEqual( + expect([...stitcher.stitch.mock.calls[0][0].entityRefs!]).toEqual( expect.arrayContaining(['k:ns/me', 'k:ns/other2']), ); await engine.stop(); @@ -633,6 +640,7 @@ describe('DefaultCatalogProcessingEngine', () => { config: new ConfigReader({}), logger: getVoidLogger(), processingDatabase: db, + knex: {} as any, orchestrator: orchestrator, stitcher: stitcher, createHash: () => hash, @@ -699,7 +707,7 @@ describe('DefaultCatalogProcessingEngine', () => { await waitForExpect(() => { expect(stitcher.stitch).toHaveBeenCalledTimes(1); }); - expect([...stitcher.stitch.mock.calls[0][0]]).toEqual( + expect([...stitcher.stitch.mock.calls[0][0].entityRefs!]).toEqual( expect.arrayContaining(['k:ns/me', 'k:ns/other2']), ); await engine.stop(); diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts index bed1fe6f50..72f92b9b40 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts @@ -22,16 +22,13 @@ import { import { assertError, serializeError, stringifyError } from '@backstage/errors'; import { Hash } from 'crypto'; import stableStringify from 'fast-json-stable-stringify'; +import { Knex } from 'knex'; import { Logger } from 'winston'; import { metrics, trace } from '@opentelemetry/api'; import { ProcessingDatabase, RefreshStateItem } from '../database/types'; import { createCounterMetric, createSummaryMetric } from '../util/metrics'; -import { - CatalogProcessingEngine, - CatalogProcessingOrchestrator, - EntityProcessingResult, -} from './types'; -import { Stitcher } from '../stitching/Stitcher'; +import { CatalogProcessingOrchestrator, EntityProcessingResult } from './types'; +import { Stitcher } from '../stitching/types'; import { startTaskPipeline } from './TaskPipeline'; import { PluginTaskScheduler } from '@backstage/backend-tasks'; import { Config } from '@backstage/config'; @@ -40,6 +37,7 @@ import { TRACER_ID, withActiveSpan, } from '../util/opentelemetry'; +import { deleteOrphanedEntities } from '../database/operations/util/deleteOrphanedEntities'; const CACHE_TTL = 5; @@ -47,10 +45,17 @@ const tracer = trace.getTracer(TRACER_ID); export type ProgressTracker = ReturnType; -export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { +// NOTE(freben): Perhaps surprisingly, this class does not implement the +// CatalogProcessingEngine type. That type is externally visible and its name is +// the way it is for historic reasons. This class has no particular reason to +// implement that precise interface; nowadays there are several different +// engines "hiding" behind the CatalogProcessingEngine interface, of which this +// is just one. +export class DefaultCatalogProcessingEngine { private readonly config: Config; private readonly scheduler?: PluginTaskScheduler; private readonly logger: Logger; + private readonly knex: Knex; private readonly processingDatabase: ProcessingDatabase; private readonly orchestrator: CatalogProcessingOrchestrator; private readonly stitcher: Stitcher; @@ -69,6 +74,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { config: Config; scheduler?: PluginTaskScheduler; logger: Logger; + knex: Knex; processingDatabase: ProcessingDatabase; orchestrator: CatalogProcessingOrchestrator; stitcher: Stitcher; @@ -84,6 +90,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { this.config = options.config; this.scheduler = options.scheduler; this.logger = options.logger; + this.knex = options.knex; this.processingDatabase = options.processingDatabase; this.orchestrator = options.orchestrator; this.stitcher = options.stitcher; @@ -255,9 +262,11 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { resultHash, }); }); - await this.stitcher.stitch( - new Set([stringifyEntityRef(unprocessedEntity)]), - ); + + await this.stitcher.stitch({ + entityRefs: [stringifyEntityRef(unprocessedEntity)], + }); + track.markSuccessfulWithErrors(); return; } @@ -305,9 +314,11 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { } }); - await this.stitcher.stitch(setOfThingsToStitch); + await this.stitcher.stitch({ + entityRefs: setOfThingsToStitch, + }); - track.markSuccessfulWithChanges(setOfThingsToStitch.size); + track.markSuccessfulWithChanges(); } catch (error) { assertError(error); track.markFailed(error); @@ -318,20 +329,21 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { } private startOrphanCleanup(): () => void { - const strategy = + const orphanStrategy = this.config.getOptionalString('catalog.orphanStrategy') ?? 'keep'; - if (strategy !== 'delete') { + if (orphanStrategy !== 'delete') { return () => {}; } const runOnce = async () => { try { - await this.processingDatabase.transaction(async tx => { - const n = await this.processingDatabase.deleteOrphanedEntities(tx); - if (n > 0) { - this.logger.info(`Deleted ${n} orphaned entities`); - } + const n = await deleteOrphanedEntities({ + knex: this.knex, + strategy: { mode: 'immediate' }, }); + if (n > 0) { + this.logger.info(`Deleted ${n} orphaned entities`); + } } catch (error) { this.logger.warn(`Failed to delete orphaned entities`, error); } @@ -363,10 +375,6 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { // Helps wrap the timing and logging behaviors function progressTracker() { // prom-client metrics are deprecated in favour of OpenTelemetry metrics. - const promStitchedEntities = createCounterMetric({ - name: 'catalog_stitched_entities_count', - help: 'Amount of entities stitched. DEPRECATED, use OpenTelemetry metrics instead', - }); const promProcessedEntities = createCounterMetric({ name: 'catalog_processed_entities_count', help: 'Amount of entities processed, DEPRECATED, use OpenTelemetry metrics instead', @@ -388,11 +396,6 @@ function progressTracker() { }); const meter = metrics.getMeter('default'); - const stitchedEntities = meter.createCounter( - 'catalog.stitched.entities.count', - { description: 'Amount of entities stitched' }, - ); - const processedEntities = meter.createCounter( 'catalog.processed.entities.count', { description: 'Amount of entities processed' }, @@ -464,13 +467,11 @@ function progressTracker() { processedEntities.add(1, { result: 'errors' }); } - function markSuccessfulWithChanges(stitchedCount: number) { + function markSuccessfulWithChanges() { endOverallTimer({ result: 'changed' }); - promStitchedEntities.inc(stitchedCount); promProcessedEntities.inc({ result: 'changed' }, 1); processingDuration.record(endTime(), { result: 'changed' }); - stitchedEntities.add(stitchedCount); processedEntities.add(1, { result: 'changed' }); } diff --git a/plugins/catalog-backend/src/processing/types.ts b/plugins/catalog-backend/src/processing/types.ts index c0d0e5db0d..002747d4c4 100644 --- a/plugins/catalog-backend/src/processing/types.ts +++ b/plugins/catalog-backend/src/processing/types.ts @@ -64,8 +64,8 @@ export interface CatalogProcessingOrchestrator { } /** - * Represents the engine that drives the processing loops. Some backend - * instances may choose to not call start, if they focus only on API + * Represents the engine that drives the processing and stitching loops. Some + * backend instances may choose to not call start, if they focus only on API * interactions. * * @public diff --git a/plugins/catalog-backend/src/service/CatalogBuilder.ts b/plugins/catalog-backend/src/service/CatalogBuilder.ts index 531bb91dd2..c12178e888 100644 --- a/plugins/catalog-backend/src/service/CatalogBuilder.ts +++ b/plugins/catalog-backend/src/service/CatalogBuilder.ts @@ -65,7 +65,7 @@ import { DefaultCatalogProcessingEngine } from '../processing/DefaultCatalogProc import { DefaultLocationService } from './DefaultLocationService'; import { DefaultEntitiesCatalog } from './DefaultEntitiesCatalog'; import { DefaultCatalogProcessingOrchestrator } from '../processing/DefaultCatalogProcessingOrchestrator'; -import { Stitcher } from '../stitching/Stitcher'; +import { DefaultStitcher } from '../stitching/DefaultStitcher'; import { createRandomProcessingInterval, ProcessingIntervalFunction, @@ -449,6 +449,11 @@ export class CatalogBuilder { await applyDatabaseMigrations(dbClient); } + const stitcher = DefaultStitcher.fromConfig(config, { + knex: dbClient, + logger, + }); + const processingDatabase = new DefaultProcessingDatabase({ database: dbClient, logger, @@ -474,7 +479,6 @@ export class CatalogBuilder { policy, legacySingleProcessorValidation: this.legacySingleProcessorValidation, }); - const stitcher = new Stitcher(dbClient, logger); const unauthorizedEntitiesCatalog = new DefaultEntitiesCatalog({ database: dbClient, logger, @@ -535,6 +539,7 @@ export class CatalogBuilder { config, scheduler, logger, + knex: dbClient, processingDatabase, orchestrator, stitcher, @@ -572,7 +577,16 @@ export class CatalogBuilder { await connectEntityProviders(providerDatabase, entityProviders); return { - processingEngine, + processingEngine: { + async start() { + await processingEngine.start(); + await stitcher.start(); + }, + async stop() { + await processingEngine.stop(); + await stitcher.stop(); + }, + }, router, }; } diff --git a/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.test.ts b/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.test.ts index 4f011813e6..fde3408c4b 100644 --- a/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.test.ts +++ b/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.test.ts @@ -30,10 +30,10 @@ import { DbRefreshStateRow, DbSearchRow, } from '../database/tables'; -import { Stitcher } from '../stitching/Stitcher'; -import { buildEntitySearch } from '../stitching/buildEntitySearch'; +import { Stitcher } from '../stitching/types'; import { DefaultEntitiesCatalog } from './DefaultEntitiesCatalog'; import { EntitiesRequest } from '../catalog/types'; +import { buildEntitySearch } from '../database/operations/stitcher/buildEntitySearch'; jest.setTimeout(60_000); @@ -1684,9 +1684,9 @@ describe('DefaultEntitiesCatalog', () => { { entity_ref: 'k:default/unrelated1', result_hash: 'not-changed' }, { entity_ref: 'k:default/unrelated2', result_hash: 'not-changed' }, ]); - expect(stitch).toHaveBeenCalledWith( - new Set(['k:default/unrelated1', 'k:default/unrelated2']), - ); + expect(stitch).toHaveBeenCalledWith({ + entityRefs: new Set(['k:default/unrelated1', 'k:default/unrelated2']), + }); }, ); }); diff --git a/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.ts b/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.ts index c328c83823..24f0432419 100644 --- a/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.ts +++ b/plugins/catalog-backend/src/service/DefaultEntitiesCatalog.ts @@ -49,8 +49,7 @@ import { DbRelationsRow, DbSearchRow, } from '../database/tables'; - -import { Stitcher } from '../stitching/Stitcher'; +import { Stitcher } from '../stitching/types'; import { isQueryEntitiesCursorRequest, @@ -606,7 +605,9 @@ export class DefaultEntitiesCatalog implements EntitiesCatalog { .where('entity_id', uid) .delete(); - await this.stitcher.stitch(new Set(relationPeers.map(p => p.ref))); + await this.stitcher.stitch({ + entityRefs: new Set(relationPeers.map(p => p.ref)), + }); } async entityAncestry(rootRef: string): Promise { diff --git a/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts b/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts index 76284ebbfd..5b29bb34d5 100644 --- a/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts +++ b/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts @@ -31,9 +31,9 @@ import { import { ProcessingDatabase } from '../database/types'; import { DefaultCatalogProcessingEngine } from '../processing/DefaultCatalogProcessingEngine'; import { EntityProcessingRequest } from '../processing/types'; -import { Stitcher } from '../stitching/Stitcher'; import { DefaultRefreshService } from './DefaultRefreshService'; import { ConfigReader } from '@backstage/config'; +import { DefaultStitcher } from '../stitching/DefaultStitcher'; jest.setTimeout(60_000); @@ -109,10 +109,16 @@ describe('DefaultRefreshService', () => { } } + const stitcher = DefaultStitcher.fromConfig(new ConfigReader({}), { + knex, + logger: defaultLogger, + }); const engine = new DefaultCatalogProcessingEngine({ config: new ConfigReader({}), logger: defaultLogger, processingDatabase: db, + knex: knex, + stitcher: stitcher, orchestrator: { async process(request: EntityProcessingRequest) { const entityRef = stringifyEntityRef(request.entity); @@ -151,7 +157,6 @@ describe('DefaultRefreshService', () => { }; }, }, - stitcher: new Stitcher(knex, defaultLogger), createHash: () => createHash('sha1'), pollingIntervalMs: 50, }); diff --git a/plugins/catalog-backend/src/stitching/Stitcher.test.ts b/plugins/catalog-backend/src/stitching/DefaultStitcher.test.ts similarity index 95% rename from plugins/catalog-backend/src/stitching/Stitcher.test.ts rename to plugins/catalog-backend/src/stitching/DefaultStitcher.test.ts index 1efcb01a3c..5b33fc424c 100644 --- a/plugins/catalog-backend/src/stitching/Stitcher.test.ts +++ b/plugins/catalog-backend/src/stitching/DefaultStitcher.test.ts @@ -25,7 +25,7 @@ import { DbRelationsRow, DbSearchRow, } from '../database/tables'; -import { Stitcher } from './Stitcher'; +import { DefaultStitcher } from './DefaultStitcher'; jest.setTimeout(60_000); @@ -41,7 +41,11 @@ describe('Stitcher', () => { const db = await databases.init(databaseId); await applyDatabaseMigrations(db); - const stitcher = new Stitcher(db, logger); + const stitcher = new DefaultStitcher({ + knex: db, + logger, + strategy: { mode: 'immediate' }, + }); let entities: DbFinalEntitiesRow[]; let entity: Entity; @@ -85,7 +89,7 @@ describe('Stitcher', () => { }, ]); - await stitcher.stitch(new Set(['k:ns/n'])); + await stitcher.stitch({ entityRefs: ['k:ns/n'] }); entities = await db('final_entities'); @@ -165,7 +169,7 @@ describe('Stitcher', () => { ); // Re-stitch without any changes - await stitcher.stitch(new Set(['k:ns/n'])); + await stitcher.stitch({ entityRefs: ['k:ns/n'] }); entities = await db('final_entities'); expect(entities.length).toBe(1); @@ -183,7 +187,7 @@ describe('Stitcher', () => { }, ]); - await stitcher.stitch(new Set(['k:ns/n'])); + await stitcher.stitch({ entityRefs: ['k:ns/n'] }); entities = await db('final_entities'); diff --git a/plugins/catalog-backend/src/stitching/DefaultStitcher.ts b/plugins/catalog-backend/src/stitching/DefaultStitcher.ts new file mode 100644 index 0000000000..095d80ad83 --- /dev/null +++ b/plugins/catalog-backend/src/stitching/DefaultStitcher.ts @@ -0,0 +1,124 @@ +/* + * Copyright 2021 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 { Config } from '@backstage/config'; +import { Knex } from 'knex'; +import splitToChunks from 'lodash/chunk'; +import { DateTime } from 'luxon'; +import { Logger } from 'winston'; +import { performStitching } from '../database/operations/stitcher/performStitching'; +import { DbRefreshStateRow } from '../database/tables'; +import { progressTracker } from './progressTracker'; +import { Stitcher, StitchingStrategy } from './types'; + +type StitchProgressTracker = ReturnType; + +/** + * Performs the act of stitching - to take all of the various outputs from the + * ingestion process, and stitching them together into the final entity JSON + * shape. + */ +export class DefaultStitcher implements Stitcher { + private readonly knex: Knex; + private readonly logger: Logger; + private readonly strategy: StitchingStrategy; + private readonly tracker: StitchProgressTracker; + + static fromConfig( + _config: Config, + options: { + knex: Knex; + logger: Logger; + }, + ): DefaultStitcher { + return new DefaultStitcher({ + knex: options.knex, + logger: options.logger, + strategy: { mode: 'immediate' }, + }); + } + + constructor(options: { + knex: Knex; + logger: Logger; + strategy: StitchingStrategy; + }) { + this.knex = options.knex; + this.logger = options.logger; + this.strategy = options.strategy; + this.tracker = progressTracker(options.knex, options.logger); + } + + async stitch(options: { + entityRefs?: Iterable; + entityIds?: Iterable; + }) { + const { entityRefs, entityIds } = options; + + if (entityRefs) { + for (const entityRef of entityRefs) { + await this.#stitchOne({ entityRef }); + } + } + + if (entityIds) { + const chunks = splitToChunks( + Array.isArray(entityIds) ? entityIds : [...entityIds], + 100, + ); + for (const chunk of chunks) { + const rows = await this.knex('refresh_state') + .select('entity_ref') + .whereIn('entity_id', chunk); + for (const row of rows) { + await this.#stitchOne({ entityRef: row.entity_ref }); + } + } + } + } + + async start() { + // Only called immediately for now + } + + async stop() { + // Only called immediately for now + } + + async #stitchOne(options: { + entityRef: string; + stitchTicket?: string; + stitchRequestedAt?: DateTime; + }) { + const track = this.tracker.stitchStart({ + entityRef: options.entityRef, + stitchRequestedAt: options.stitchRequestedAt, + }); + + try { + const result = await performStitching({ + knex: this.knex, + logger: this.logger, + strategy: this.strategy, + entityRef: options.entityRef, + stitchTicket: options.stitchTicket, + }); + track.markComplete(result); + } catch (error) { + track.markFailed(error); + } + } +} diff --git a/plugins/catalog-backend/src/stitching/Stitcher.ts b/plugins/catalog-backend/src/stitching/Stitcher.ts deleted file mode 100644 index 38d0fd263f..0000000000 --- a/plugins/catalog-backend/src/stitching/Stitcher.ts +++ /dev/null @@ -1,250 +0,0 @@ -/* - * Copyright 2021 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 { ENTITY_STATUS_CATALOG_PROCESSING_TYPE } from '@backstage/catalog-client'; -import { AlphaEntity, EntityStatusItem } from '@backstage/catalog-model/alpha'; -import { - ANNOTATION_EDIT_URL, - ANNOTATION_VIEW_URL, - EntityRelation, -} from '@backstage/catalog-model'; -import { SerializedError, stringifyError } from '@backstage/errors'; -import { Knex } from 'knex'; -import { v4 as uuid } from 'uuid'; -import { Logger } from 'winston'; -import { - DbFinalEntitiesRow, - DbRefreshStateRow, - DbSearchRow, -} from '../database/tables'; -import { buildEntitySearch } from './buildEntitySearch'; -import { BATCH_SIZE, generateStableHash } from './util'; - -// See https://github.com/facebook/react/blob/f0cf832e1d0c8544c36aa8b310960885a11a847c/packages/react-dom-bindings/src/shared/sanitizeURL.js -const scriptProtocolPattern = - // eslint-disable-next-line no-control-regex - /^[\u0000-\u001F ]*j[\r\n\t]*a[\r\n\t]*v[\r\n\t]*a[\r\n\t]*s[\r\n\t]*c[\r\n\t]*r[\r\n\t]*i[\r\n\t]*p[\r\n\t]*t[\r\n\t]*\:/i; - -/** - * Performs the act of stitching - to take all of the various outputs from the - * ingestion process, and stitching them together into the final entity JSON - * shape. - */ -export class Stitcher { - constructor( - private readonly database: Knex, - private readonly logger: Logger, - ) {} - - async stitch(entityRefs: Set) { - for (const entityRef of entityRefs) { - try { - await this.stitchOne(entityRef); - } catch (error) { - this.logger.error( - `Failed to stitch ${entityRef}, ${stringifyError(error)}`, - ); - } - } - } - - private async stitchOne(entityRef: string): Promise { - const entityResult = await this.database('refresh_state') - .where({ entity_ref: entityRef }) - .limit(1) - .select('entity_id'); - if (!entityResult.length) { - // Entity does no exist in refresh state table, no stitching required. - return; - } - - // Insert stitching ticket that will be compared before inserting the final entity. - const ticket = uuid(); - await this.database('final_entities') - .insert({ - entity_id: entityResult[0].entity_id, - hash: '', - stitch_ticket: ticket, - }) - .onConflict('entity_id') - .merge(['stitch_ticket']); - - // Selecting from refresh_state and final_entities should yield exactly - // one row (except in abnormal cases where the stitch was invoked for - // something that didn't exist at all, in which case it's zero rows). - // The join with the temporary incoming_references still gives one row. - const [processedResult, relationsResult] = await Promise.all([ - this.database - .with('incoming_references', function incomingReferences(builder) { - return builder - .from('refresh_state_references') - .where({ target_entity_ref: entityRef }) - .count({ count: '*' }); - }) - .select({ - entityId: 'refresh_state.entity_id', - processedEntity: 'refresh_state.processed_entity', - errors: 'refresh_state.errors', - incomingReferenceCount: 'incoming_references.count', - previousHash: 'final_entities.hash', - }) - .from('refresh_state') - .where({ 'refresh_state.entity_ref': entityRef }) - .crossJoin(this.database.raw('incoming_references')) - .leftOuterJoin('final_entities', { - 'final_entities.entity_id': 'refresh_state.entity_id', - }), - this.database - .distinct({ - relationType: 'type', - relationTarget: 'target_entity_ref', - }) - .from('relations') - .where({ source_entity_ref: entityRef }) - .orderBy('relationType', 'asc') - .orderBy('relationTarget', 'asc'), - ]); - - // If there were no rows returned, it would mean that there was no - // matching row even in the refresh_state. This can happen for example - // if we emit a relation to something that hasn't been ingested yet. - // It's safe to ignore this stitch attempt in that case. - if (!processedResult.length) { - this.logger.error( - `Unable to stitch ${entityRef}, item does not exist in refresh state table`, - ); - return; - } - - const { - entityId, - processedEntity, - errors, - incomingReferenceCount, - previousHash, - } = processedResult[0]; - - // If there was no processed entity in place, the target hasn't been - // through the processing steps yet. It's safe to ignore this stitch - // attempt in that case, since another stitch will be triggered when - // that processing has finished. - if (!processedEntity) { - this.logger.debug( - `Unable to stitch ${entityRef}, the entity has not yet been processed`, - ); - return; - } - - // Grab the processed entity and stitch all of the relevant data into - // it - const entity = JSON.parse(processedEntity) as AlphaEntity; - const isOrphan = Number(incomingReferenceCount) === 0; - let statusItems: EntityStatusItem[] = []; - - if (isOrphan) { - this.logger.debug(`${entityRef} is an orphan`); - entity.metadata.annotations = { - ...entity.metadata.annotations, - ['backstage.io/orphan']: 'true', - }; - } - if (errors) { - const parsedErrors = JSON.parse(errors) as SerializedError[]; - if (Array.isArray(parsedErrors) && parsedErrors.length) { - statusItems = parsedErrors.map(e => ({ - type: ENTITY_STATUS_CATALOG_PROCESSING_TYPE, - level: 'error', - message: `${e.name}: ${e.message}`, - error: e, - })); - } - } - // We opt to do this check here as we otherwise can't guarantee that it will be run after all processors - for (const annotation of [ANNOTATION_VIEW_URL, ANNOTATION_EDIT_URL]) { - const value = entity.metadata.annotations?.[annotation]; - if (typeof value === 'string' && scriptProtocolPattern.test(value)) { - entity.metadata.annotations![annotation] = - 'https://backstage.io/annotation-rejected-for-security-reasons'; - } - } - - // TODO: entityRef is lower case and should be uppercase in the final - // result - entity.relations = relationsResult - .filter(row => row.relationType /* exclude null row, if relevant */) - .map(row => ({ - type: row.relationType!, - targetRef: row.relationTarget!, - })); - if (statusItems.length) { - entity.status = { - ...entity.status, - items: [...(entity.status?.items ?? []), ...statusItems], - }; - } - - // If the output entity was actually not changed, just abort - const hash = generateStableHash(entity); - if (hash === previousHash) { - this.logger.debug(`Skipped stitching of ${entityRef}, no changes`); - return; - } - - entity.metadata.uid = entityId; - if (!entity.metadata.etag) { - // If the original data source did not have its own etag handling, - // use the hash as a good-quality etag - entity.metadata.etag = hash; - } - - // This may throw if the entity is invalid, so we call it before - // the final_entities write, even though we may end up not needing - // to write the search index. - const searchEntries = buildEntitySearch(entityId, entity); - - const amountOfRowsChanged = await this.database( - 'final_entities', - ) - .update({ - final_entity: JSON.stringify(entity), - hash, - last_updated_at: this.database.fn.now(), - }) - .where('entity_id', entityId) - .where('stitch_ticket', ticket) - .onConflict('entity_id') - .merge(['final_entity', 'hash', 'last_updated_at']); - - if (amountOfRowsChanged === 0) { - this.logger.debug( - `Entity ${entityRef} is already processed, skipping write.`, - ); - return; - } - - // TODO(freben): Search will probably need a similar safeguard against - // race conditions like the final_entities ticket handling above. - // Otherwise, it can be the case that: - // A writes the entity -> - // B writes the entity -> - // B writes search -> - // A writes search - await this.database('search') - .where({ entity_id: entityId }) - .delete(); - await this.database.batchInsert('search', searchEntries, BATCH_SIZE); - } -} diff --git a/plugins/catalog-backend/src/stitching/progressTracker.ts b/plugins/catalog-backend/src/stitching/progressTracker.ts new file mode 100644 index 0000000000..e8e08dbc54 --- /dev/null +++ b/plugins/catalog-backend/src/stitching/progressTracker.ts @@ -0,0 +1,84 @@ +/* + * 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 { stringifyError } from '@backstage/errors'; +import { metrics } from '@opentelemetry/api'; +import { Knex } from 'knex'; +import { DateTime } from 'luxon'; +import { Logger } from 'winston'; +import { createCounterMetric } from '../util/metrics'; + +// Helps wrap the timing and logging behaviors +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', + help: 'Amount of entities stitched. DEPRECATED, use OpenTelemetry metrics instead', + }); + + const meter = metrics.getMeter('default'); + + const stitchedEntities = meter.createCounter( + 'catalog.stitched.entities.count', + { + description: 'Amount of entities stitched', + }, + ); + + const stitchingDuration = meter.createHistogram( + 'catalog.stitching.duration', + { + description: 'Time spent executing the full stitching flow', + unit: 'seconds', + }, + ); + + function stitchStart(item: { + entityRef: string; + stitchRequestedAt?: DateTime; + }) { + logger.debug(`Stitching ${item.entityRef}`); + + const startTime = process.hrtime(); + + function endTime() { + const delta = process.hrtime(startTime); + return delta[0] + delta[1] / 1e9; + } + + function markComplete(result: string) { + promStitchedEntities.inc(1); + stitchedEntities.add(1, { result }); + stitchingDuration.record(endTime(), { result }); + } + + function markFailed(error: Error) { + promStitchedEntities.inc(1); + stitchedEntities.add(1, { result: 'error' }); + stitchingDuration.record(endTime(), { result: 'error' }); + logger.error( + `Failed to stitch ${item.entityRef}, ${stringifyError(error)}`, + ); + } + + return { + markComplete, + markFailed, + }; + } + + return { stitchStart }; +} diff --git a/plugins/catalog-backend/src/stitching/types.ts b/plugins/catalog-backend/src/stitching/types.ts new file mode 100644 index 0000000000..5f32a85b1f --- /dev/null +++ b/plugins/catalog-backend/src/stitching/types.ts @@ -0,0 +1,40 @@ +/* + * 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. + */ + +/** + * Performs the act of stitching - to take all of the various outputs from the + * ingestion process, and stitching them together into the final entity JSON + * shape. + */ +export interface Stitcher { + stitch(options: { + entityRefs?: Iterable; + entityIds?: Iterable; + }): Promise; +} + +/** + * The strategies supported by the stitching process, in terms of when to + * perform stitching. + * + * @remarks + * + * In immediate mode, stitching happens "in-band" (blocking) immediately when + * each processing task finishes. + */ +export type StitchingStrategy = { + mode: 'immediate'; +}; diff --git a/plugins/catalog-backend/src/tests/integration.test.ts b/plugins/catalog-backend/src/tests/integration.test.ts index e7f983f51f..8aeb95f49b 100644 --- a/plugins/catalog-backend/src/tests/integration.test.ts +++ b/plugins/catalog-backend/src/tests/integration.test.ts @@ -53,7 +53,7 @@ import { CatalogProcessingEngine } from '../processing/types'; import { DefaultEntitiesCatalog } from '../service/DefaultEntitiesCatalog'; import { DefaultRefreshService } from '../service/DefaultRefreshService'; import { RefreshOptions, RefreshService } from '../service/types'; -import { Stitcher } from '../stitching/Stitcher'; +import { DefaultStitcher } from '../stitching/DefaultStitcher'; const voidLogger = getVoidLogger(); @@ -268,7 +268,7 @@ class TestHarness { policy: EntityPolicies.allOf([]), legacySingleProcessorValidation: false, }); - const stitcher = new Stitcher(db, logger); + const stitcher = DefaultStitcher.fromConfig(config, { knex: db, logger }); const catalog = new DefaultEntitiesCatalog({ database: db, logger, @@ -282,6 +282,7 @@ class TestHarness { config: new ConfigReader({}), logger, processingDatabase, + knex: db, orchestrator, stitcher, createHash: () => createHash('sha1'), @@ -300,7 +301,16 @@ class TestHarness { return new TestHarness( catalog, - engine, + { + async start() { + await engine.start(); + await stitcher.start(); + }, + async stop() { + await engine.stop(); + await stitcher.stop(); + }, + }, refresh, provider, proxyProgressTracker, diff --git a/plugins/catalog-backend/src/tests/performance/stitchingPerformance.test.ts b/plugins/catalog-backend/src/tests/performance/stitchingPerformance.test.ts index e60c0a33d6..c1a869f9ad 100644 --- a/plugins/catalog-backend/src/tests/performance/stitchingPerformance.test.ts +++ b/plugins/catalog-backend/src/tests/performance/stitchingPerformance.test.ts @@ -181,6 +181,7 @@ describePerformanceTest('stitchingPerformance', () => { const backend = await startTestBackend({ features: [ import('@backstage/plugin-catalog-backend/alpha'), + staticDatabase(knex), createBackendModule({ moduleId: 'syntheticLoadEntities', pluginId: 'catalog', @@ -200,7 +201,6 @@ describePerformanceTest('stitchingPerformance', () => { }); }, }), - staticDatabase(knex), ], }); From e34919037404548f994dbe113acb908f6f0975aa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Tue, 26 Sep 2023 10:41:14 +0200 Subject: [PATCH 2/2] update changeset MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- .changeset/polite-donuts-fail.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/polite-donuts-fail.md b/.changeset/polite-donuts-fail.md index 4b71dfddc3..9c03faaf44 100644 --- a/.changeset/polite-donuts-fail.md +++ b/.changeset/polite-donuts-fail.md @@ -2,4 +2,4 @@ '@backstage/plugin-catalog-backend': patch --- -Internal refactors as a path toward #18062 +Internal refactors, laying the foundation for later introducing deferred stitching (see #18062).