Add to contrib: ImmediateEntityProvider, LoadTestingEntityProvider
Signed-off-by: Fredrik Adelöw <freben@gmail.com>
This commit is contained in:
Fredrik Adelöw
@@ -0,0 +1,5 @@
|
||||
---
|
||||
'@backstage/plugin-catalog-backend': patch
|
||||
---
|
||||
|
||||
Enable internal batching of very large deletions, to not run into SQL binding limits
|
||||
@@ -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>;
|
||||
}
|
||||
@@ -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>;
|
||||
}
|
||||
@@ -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)}`,
|
||||
|
||||
Reference in New Issue
Block a user