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

Merge pull request #10325 from backstage/freben/providers

Browse Source 
Add to contrib: `ImmediateEntityProvider`, `LoadTestingEntityProvider`
This commit is contained in:
Fredrik Adelöw
2022-03-21 14:28:11 +01:00
committed by GitHub
parent 2651fa8db4 1cb55c667e
commit 43d323c917
5 changed files with 427 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/modern-pumas-join.md
+5
View File
@@ -0,0 +1,5 @@
---
'@backstage/plugin-catalog-backend': patch
---
Enable internal batching of very large deletions, to not run into SQL binding limits
contrib/catalog/ImmediateEntityProvider.ts
+169
View File
@@ -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<DeferredEntity[]> {
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<void>;
}
contrib/catalog/LoadTestingEntityProvider.ts
+151
View File
@@ -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<void>;
}
contrib/catalog/README.md
+27
View File
@@ -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.
plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts
+75 -73
View File
@@ -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<DbRefreshStateRow>('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<DbRefreshStateRow>('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<DbRefreshStateReferencesRow>('refresh_state_references')
.where('source_key', '=', options.sourceKey)
.whereIn('target_entity_ref', toRemove)
.delete();
await tx<DbRefreshStateReferencesRow>('refresh_state_references')
.where('source_key', '=', options.sourceKey)
.whereIn('target_entity_ref', refs)
.delete();
}
this.options.logger.debug(
`removed, ${removedCount} entities: ${JSON.stringify(toRemove)}`,