From 95408dbe99cb78044726da8240809fdc5eaef4a1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Fri, 18 Mar 2022 11:41:30 +0100 Subject: [PATCH 1/2] Add to contrib: `ImmediateEntityProvider`, `LoadTestingEntityProvider` MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- .changeset/modern-pumas-join.md | 5 + contrib/catalog/ImmediateEntityProvider.ts | 169 ++++++++++++++++++ contrib/catalog/LoadTestingEntityProvider.ts | 151 ++++++++++++++++ .../src/database/DefaultProcessingDatabase.ts | 148 +++++++-------- 4 files changed, 400 insertions(+), 73 deletions(-) create mode 100644 .changeset/modern-pumas-join.md create mode 100644 contrib/catalog/ImmediateEntityProvider.ts create mode 100644 contrib/catalog/LoadTestingEntityProvider.ts diff --git a/.changeset/modern-pumas-join.md b/.changeset/modern-pumas-join.md new file mode 100644 index 0000000000..d2995af986 --- /dev/null +++ b/.changeset/modern-pumas-join.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend': patch +--- + +Enable internal batching of very large deletions, to not run into SQL binding limits diff --git a/contrib/catalog/ImmediateEntityProvider.ts b/contrib/catalog/ImmediateEntityProvider.ts new file mode 100644 index 0000000000..b66c62e330 --- /dev/null +++ b/contrib/catalog/ImmediateEntityProvider.ts @@ -0,0 +1,169 @@ +import { + ANNOTATION_LOCATION, + ANNOTATION_ORIGIN_LOCATION, + Entity, + entitySchemaValidator, +} from '@backstage/catalog-model'; +import { InputError } from '@backstage/errors'; +import { + DeferredEntity, + EntityProvider, + EntityProviderConnection, + parseEntityYaml, +} from '@backstage/plugin-catalog-backend'; +import bodyParser from 'body-parser'; +import express from 'express'; +import Router from 'express-promise-router'; +import lodash from 'lodash'; +import { Logger } from 'winston'; + +/** + * An entity provider attached to a router, that lets users perform direct + * manipulation of a set of entities using REST requests. + * + * @remarks + * + * Installation: + * + * Add it to the catalog builder in your + * `packages/backend/src/plugins/catalog.ts`. Note that it BOTH adds a provider + * and amends the catalog router: + * + * ``` + * const immediate = new ImmediateEntityProvider({ + * logger: env.logger, + * handleEntity: (deferred) => { + * // Optionally modify the incoming entity + * }, + * }); + * builder.addEntityProvider(immediate); + * + * // ... + * + * return router.use('/immediate', immediate.getRouter()); + * ``` + * + * API (assume a catalog prefix, e.g. `/api/catalog`): + * + * - `POST /immediate/entities`: Accepts a YAML document of entities, and + * inserts or updates the entities that match that document. Returns 201 OK on + * success. + * + * - `PUT /immediate/entities`: Accepts a YAML document of entities, and + * replaces the entire set of entities managed by the provider with those + * entities. Returns 201 OK on success. + */ +export class ImmediateEntityProvider implements EntityProvider { + private connection?: EntityProviderConnection; + private readonly entityValidator: (data: unknown) => Entity; + + constructor(private readonly options: ImmediateEntityProviderOptions) { + this.entityValidator = entitySchemaValidator(); + } + + /** {@inheritdoc @backstage/plugin-catalog-backend#EntityProvider.getProviderName} */ + getProviderName() { + return `ImmediateEntityProvider`; + } + + /** {@inheritdoc @backstage/plugin-catalog-backend#EntityProvider.connect} */ + async connect(connection: EntityProviderConnection) { + this.connection = connection; + } + + getRouter(): express.Router { + const router = Router(); + + router.use(bodyParser.raw({ type: '*/*' })); + + router.post('/entities', async (req, res) => { + if (!this.connection) { + throw new Error(`Service is not yet initialized`); + } + const deferred = await this.getRequestBodyEntities(req); + await this.connection.applyMutation({ + type: 'delta', + added: deferred, + removed: [], + }); + res.status(201).end(); + }); + + router.put('/entities', async (req, res) => { + if (!this.connection) { + throw new Error(`Service is not yet initialized`); + } + const deferred = await this.getRequestBodyEntities(req); + await this.connection.applyMutation({ + type: 'full', + entities: deferred, + }); + res.status(201).end(); + }); + + return router; + } + + private async getRequestBodyEntities( + req: express.Request, + ): Promise { + if (!Buffer.isBuffer(req.body) || !req.body.length) { + throw new InputError(`Missing request body`); + } + + const result: DeferredEntity[] = []; + + for await (const item of parseEntityYaml(req.body, { + type: 'immediate', + target: 'immediate', + })) { + if (item.type === 'entity') { + const deferred: DeferredEntity = { + entity: lodash.merge( + { + metadata: { + annotations: { + [ANNOTATION_ORIGIN_LOCATION]: 'immediate:immediate', + [ANNOTATION_LOCATION]: 'immediate:immediate', + }, + }, + }, + item.entity, + ), + locationKey: `immediate:`, + }; + + await this.options.handleEntity?.(req, deferred); + deferred.entity = this.entityValidator(deferred.entity); + + result.push(deferred); + } else if (item.type === 'error') { + throw new InputError(`Malformed entity YAML, ${item.error}`); + } else { + throw new InputError(`Internal error, failed to parse entity`); + } + } + + return result; + } +} + +/** + * Options for {@link ImmediateEntityProvider}. + */ +export interface ImmediateEntityProviderOptions { + /** + * The logger to use. + */ + logger: Logger; + + /** + * An optional function to perform adjustments to, or validate, an incoming + * entity before being stored. It is permitted to modify the deferred entity, + * but the request is static and has had its body consumed. + */ + handleEntity?: ( + request: express.Request, + deferred: DeferredEntity, + ) => void | Promise; +} diff --git a/contrib/catalog/LoadTestingEntityProvider.ts b/contrib/catalog/LoadTestingEntityProvider.ts new file mode 100644 index 0000000000..598099fce9 --- /dev/null +++ b/contrib/catalog/LoadTestingEntityProvider.ts @@ -0,0 +1,151 @@ +import { + ANNOTATION_LOCATION, + ANNOTATION_ORIGIN_LOCATION, + Entity, +} from '@backstage/catalog-model'; +import { + EntityProvider, + EntityProviderConnection, +} from '@backstage/plugin-catalog-backend'; +import { Logger } from 'winston'; + +/** + * An entity provider that can be used for load testing. Not for production use. + * + * @remarks + * + * Add it to the catalog builder in your + * `packages/backend/src/plugins/catalog.ts` doing some type of work, for + * example: + * + * ``` + * builder.addEntityProvider( + * new LoadTestingEntityProvider({ + * logger: env.logger, + * onStartup: async ({ connection, generateRandomEntities }) => { + * await connection.applyMutation({ + * type: 'full', + * entities: generateRandomEntities(100000).map(e => ({ + * entity: e, + * locationKey: 'l', + * })), + * }); + * }, + * }), + * ); + * ``` + * + * The provider will run the test, outputting some timing info onto the console. + * It will also clean up everything you added through the given connection. + */ +export class LoadTestingEntityProvider implements EntityProvider { + constructor(private readonly options: LoadTestingEntityProviderOptions) {} + + /** {@inheritdoc @backstage/plugin-catalog-backend#EntityProvider.getProviderName} */ + getProviderName() { + return `LoadTestingEntityProvider`; + } + + /** {@inheritdoc @backstage/plugin-catalog-backend#EntityProvider.connect} */ + async connect(connection: EntityProviderConnection) { + const delayStartup = this.options.delayStartup ?? 10_000; + const logger = this.options.logger.child({ + class: LoadTestingEntityProvider.prototype.constructor.name, + }); + + if (delayStartup) { + logger.info( + `[LOAD-TEST] Starting in ${(delayStartup / 1000).toFixed(1)}s`, + ); + } + + setTimeout(async () => { + const timer = () => { + const startedOn = Date.now(); + return () => `${((Date.now() - startedOn) / 1000).toFixed(1)}s`; + }; + + const overallTimer = timer(); + logger.info(`[LOAD-TEST] Started`); + + const runTimer = timer(); + try { + await this.options.onStartup({ + connection, + logger, + generateRandomEntities, + }); + logger.info(`[LOAD-TEST] Finished in ${runTimer()}`); + } catch (error) { + logger.error(`[LOAD-TEST] Failed after ${runTimer()}`, error); + } + + const cleanupTimer = timer(); + logger.info(`[LOAD-TEST] Running cleanup`); + await connection.applyMutation({ + type: 'full', + entities: [], + }); + + logger.info(`[LOAD-TEST] ***************************************`); + logger.info(`[LOAD-TEST] Test run time: ${runTimer()}`); + logger.info(`[LOAD-TEST] Cleanup run time: ${cleanupTimer()}`); + logger.info(`[LOAD-TEST] Total time: ${overallTimer()}`); + logger.info(`[LOAD-TEST] ***************************************`); + }, delayStartup); + } +} + +function generateRandomEntities(count: number): Entity[] { + const result: Entity[] = []; + + for (let i = 1; i <= count; ++i) { + result.push({ + apiVersion: 'backstage.io/v1alpha1', + kind: 'Component', + metadata: { + annotations: { + [ANNOTATION_ORIGIN_LOCATION]: 'url:http://example.com/load-testing', + [ANNOTATION_LOCATION]: 'url:http://example.com/load-testing', + }, + namespace: 'load-test', + name: `load-test-${i}`, + }, + spec: { + type: 'load-test-data', + owner: 'me', + lifecycle: 'experimental', + }, + }); + } + + return result; +} + +/** + * Options for LoadTestingEntityProvider. + */ +export interface LoadTestingEntityProviderOptions { + /** + * The logger to use. + */ + logger: Logger; + + /** + * The number of milliseconds of delay to wait before starting the test. This + * gives the backend a chance to settle into a stable state before the test + * starts. + * + * @defaultValue 5000 + */ + delayStartup?: number; + + /** + * What work to do on startup. + */ + onStartup: (context: { + connection: EntityProviderConnection; + logger: Logger; + generateRandomEntities(count: number): Entity[]; + }) => Promise; +} diff --git a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts index 061be0a767..c9fe0da0a3 100644 --- a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts @@ -164,8 +164,9 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { const { toAdd, toUpsert, toRemove } = await this.createDelta(tx, options); if (toRemove.length) { - // TODO(freben): Batch split, to not hit variable limits? - /* + let removedCount = 0; + for (const refs of lodash.chunk(toRemove, 1000)) { + /* WITH RECURSIVE -- All the nodes that can be reached downwards from our root descendants(root_id, entity_ref) AS ( @@ -200,78 +201,79 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { -- Exclude all lines that had such a foreign ancestor WHERE ancestors.root_id IS NULL; */ - const removedCount = await tx('refresh_state') - .whereIn('entity_ref', function orphanedEntityRefs(orphans) { - return ( - orphans - // All the nodes that can be reached downwards from our root - .withRecursive('descendants', function descendants(outer) { - return outer - .select({ root_id: 'id', entity_ref: 'target_entity_ref' }) - .from('refresh_state_references') - .where('source_key', options.sourceKey) - .whereIn('target_entity_ref', toRemove) - .union(function recursive(inner) { - return inner - .select({ - root_id: 'descendants.root_id', - entity_ref: - 'refresh_state_references.target_entity_ref', - }) - .from('descendants') - .join('refresh_state_references', { - 'descendants.entity_ref': - 'refresh_state_references.source_entity_ref', - }); - }); - }) - // All the nodes that can be reached upwards from the descendants - .withRecursive('ancestors', function ancestors(outer) { - return outer - .select({ - root_id: tx.raw('CAST(NULL as INT)', []), - via_entity_ref: 'entity_ref', - to_entity_ref: 'entity_ref', - }) - .from('descendants') - .union(function recursive(inner) { - return inner - .select({ - root_id: tx.raw( - 'CASE WHEN source_key IS NOT NULL THEN id ELSE NULL END', - [], - ), - via_entity_ref: 'source_entity_ref', - to_entity_ref: 'ancestors.to_entity_ref', - }) - .from('ancestors') - .join('refresh_state_references', { - target_entity_ref: 'ancestors.via_entity_ref', - }); - }); - }) - // Start out with all of the descendants - .select('descendants.entity_ref') - .from('descendants') - // Expand with all ancestors that point to those, but aren't the current root - .leftOuterJoin('ancestors', function keepaliveRoots() { - this.on( - 'ancestors.to_entity_ref', - '=', - 'descendants.entity_ref', - ); - this.andOnNotNull('ancestors.root_id'); - this.andOn('ancestors.root_id', '!=', 'descendants.root_id'); - }) - .whereNull('ancestors.root_id') - ); - }) - .delete(); + removedCount += await tx('refresh_state') + .whereIn('entity_ref', function orphanedEntityRefs(orphans) { + return ( + orphans + // All the nodes that can be reached downwards from our root + .withRecursive('descendants', function descendants(outer) { + return outer + .select({ root_id: 'id', entity_ref: 'target_entity_ref' }) + .from('refresh_state_references') + .where('source_key', options.sourceKey) + .whereIn('target_entity_ref', refs) + .union(function recursive(inner) { + return inner + .select({ + root_id: 'descendants.root_id', + entity_ref: + 'refresh_state_references.target_entity_ref', + }) + .from('descendants') + .join('refresh_state_references', { + 'descendants.entity_ref': + 'refresh_state_references.source_entity_ref', + }); + }); + }) + // All the nodes that can be reached upwards from the descendants + .withRecursive('ancestors', function ancestors(outer) { + return outer + .select({ + root_id: tx.raw('CAST(NULL as INT)', []), + via_entity_ref: 'entity_ref', + to_entity_ref: 'entity_ref', + }) + .from('descendants') + .union(function recursive(inner) { + return inner + .select({ + root_id: tx.raw( + 'CASE WHEN source_key IS NOT NULL THEN id ELSE NULL END', + [], + ), + via_entity_ref: 'source_entity_ref', + to_entity_ref: 'ancestors.to_entity_ref', + }) + .from('ancestors') + .join('refresh_state_references', { + target_entity_ref: 'ancestors.via_entity_ref', + }); + }); + }) + // Start out with all of the descendants + .select('descendants.entity_ref') + .from('descendants') + // Expand with all ancestors that point to those, but aren't the current root + .leftOuterJoin('ancestors', function keepaliveRoots() { + this.on( + 'ancestors.to_entity_ref', + '=', + 'descendants.entity_ref', + ); + this.andOnNotNull('ancestors.root_id'); + this.andOn('ancestors.root_id', '!=', 'descendants.root_id'); + }) + .whereNull('ancestors.root_id') + ); + }) + .delete(); - await tx('refresh_state_references') - .where('source_key', '=', options.sourceKey) - .whereIn('target_entity_ref', toRemove) - .delete(); + await tx('refresh_state_references') + .where('source_key', '=', options.sourceKey) + .whereIn('target_entity_ref', refs) + .delete(); + } this.options.logger.debug( `removed, ${removedCount} entities: ${JSON.stringify(toRemove)}`, From 1cb55c667ec63d1688baf52750057f057870c601 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Mon, 21 Mar 2022 14:12:28 +0100 Subject: [PATCH 2/2] add readme MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- contrib/catalog/README.md | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) create mode 100644 contrib/catalog/README.md diff --git a/contrib/catalog/README.md b/contrib/catalog/README.md new file mode 100644 index 0000000000..e02760cdb0 --- /dev/null +++ b/contrib/catalog/README.md @@ -0,0 +1,27 @@ +# Catalog Contrib + +This directory contains various community contributions related to [the Backstage catalog](https://backstage.io/docs/features/software-catalog/software-catalog-overview). + +There is no guarantee of correctness or fitness of purpose of these +contributions, but we hope that they are helpful to someone! + +Installation instructions are generally in the doc comment on top of each class. + +## ImmediateEntityProvider + +Sometimes we get requests for the ability to POST/PUT entities directly to the +catalog, instead of its regular mode of operation where it pulls data from +authoritative sources itself. + +The core product does not intend to support this use case, since it comes with a +number of caveats. However, this entity provider demonstrates how to build a +very basic version of such functionality yourself. It does not offer any +protection from misuse, but can serve as a good starting point to build out such +a provider yourself, fit for your particular needs. + +## LoadTestingEntityProvider + +This is a trivial little test bed entity provider that lets you make huge batch +operations and get some timings back. It also tries to clean up after itself +when it's done. This can be useful if you are working on optimizing the catalog +itself, or on processors or similar that you add to it.