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

Merge pull request #15154 from dekoding/dekoding/deltas-in-incremental-providers

Browse Source 
Add events integration to Incremental Entity Providers
This commit is contained in:
Fredrik Adelöw
2023-01-20 13:33:28 +01:00
committed by GitHub
parent c223b8194d 3c1eae9324
commit 27732dcf37
10 changed files with 356 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/bright-eagles-love.md
+5
View File
@@ -0,0 +1,5 @@
---
'@backstage/plugin-catalog-backend-module-incremental-ingestion': minor
---
Added new mechanism to handle deltas in incremental providers
plugins/catalog-backend-module-incremental-ingestion/api-report.md
+12
View File
@@ -10,6 +10,7 @@ import { CatalogBuilder } from '@backstage/plugin-catalog-backend';
import type { Config } from '@backstage/config';
import type { DeferredEntity } from '@backstage/plugin-catalog-backend';
import type { DurationObjectUnits } from 'luxon';
import { EventParams } from '@backstage/plugin-events-node';
import type { Logger } from 'winston';
import type { PermissionEvaluator } from '@backstage/plugin-permission-common';
import type { PluginDatabaseManager } from '@backstage/backend-common';
@@ -50,6 +51,17 @@ export class IncrementalCatalogBuilder {
// @public
export interface IncrementalEntityProvider<TCursor, TContext> {
around(burst: (context: TContext) => Promise<void>): Promise<void>;
eventHandler?: {
onEvent: (params: EventParams) =>
| undefined
| {
added: DeferredEntity[];
removed: {
entityRef: string;
}[];
};
supportsEventTopics: () => string[];
};
getProviderName(): string;
next(
context: TContext,
plugins/catalog-backend-module-incremental-ingestion/package.json
+1
View File
@@ -42,6 +42,7 @@
"@backstage/errors": "workspace:^",
"@backstage/plugin-catalog-backend": "workspace:^",
"@backstage/plugin-catalog-node": "workspace:^",
"@backstage/plugin-events-node": "workspace:^",
"@backstage/plugin-permission-common": "workspace:^",
"@types/express": "^4.17.6",
"@types/luxon": "^3.0.0",
plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts
+26
View File
@@ -269,6 +269,17 @@ export class IncrementalIngestionDatabaseManager {
});
}
/**
* This method is used to remove entity records from the ingestion_mark_entities
* table by their entity reference.
*/
async deleteEntityRecordsByRef(entities: { entityRef: string }[]) {
const refs = entities.map(e => e.entityRef);
await this.client.transaction(async tx => {
await tx('ingestion_mark_entities').delete().whereIn('ref', refs);
});
}
/**
* Creates a new ingestion record.
* @param provider - string
@@ -535,6 +546,21 @@ export class IncrementalIngestionDatabaseManager {
});
}
/**
* Returns the first record from `ingestion_marks` for the supplied ingestionId.
* @param ingestionId - string
* @returns MarkRecord | undefined
*/
async getFirstMark(ingestionId: string) {
return await this.client.transaction(async tx => {
const mark = await tx<MarkRecord>('ingestion_marks')
.where('ingestion_id', ingestionId)
.orderBy('sequence', 'asc')
.first();
return mark;
});
}
async getAllMarks(ingestionId: string) {
return await this.client.transaction(async tx => {
const marks = await tx<MarkRecord>('ingestion_marks')
plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts
+71 -1
View File
@@ -21,8 +21,11 @@ import { performance } from 'perf_hooks';
import { Duration, DurationObjectUnits } from 'luxon';
import { v4 } from 'uuid';
import { stringifyError } from '@backstage/errors';
import { EventParams, EventSubscriber } from '@backstage/plugin-events-node';
export class IncrementalIngestionEngine implements IterationEngine {
export class IncrementalIngestionEngine
implements IterationEngine, EventSubscriber
{
private readonly restLength: Duration;
private readonly backoff: DurationObjectUnits[];
@@ -329,4 +332,71 @@ export class IncrementalIngestionEngine implements IterationEngine {
removed,
});
}
async onEvent(params: EventParams): Promise<void> {
const { topic } = params;
if (!this.supportsEventTopics().includes(topic)) {
return;
}
const { logger, provider, connection } = this.options;
const providerName = provider.getProviderName();
logger.debug(`incremental-engine: ${providerName} received ${topic} event`);
if (!provider.eventHandler) {
return;
}
const delta = provider.eventHandler.onEvent(params);
if (delta) {
if (delta.added.length > 0) {
const ingestionRecord = await this.manager.getCurrentIngestionRecord(
providerName,
);
if (!ingestionRecord) {
logger.debug(
`incremental-engine: ${providerName} skipping delta addition because incremental ingestion is restarting.`,
);
} else {
const mark =
ingestionRecord.status === 'resting'
? await this.manager.getLastMark(ingestionRecord.id)
: await this.manager.getFirstMark(ingestionRecord.id);
if (!mark) {
throw new Error(
`Cannot apply delta, page records are missing! Please re-run incremental ingestion for ${providerName}.`,
);
}
await this.manager.createMarkEntities(mark.id, delta.added);
}
}
if (delta.removed.length > 0) {
await this.manager.deleteEntityRecordsByRef(delta.removed);
}
await connection.applyMutation({
type: 'delta',
...delta,
});
logger.debug(
`incremental-engine: ${providerName} processed delta from '${topic}' event`,
);
} else {
logger.warn(
`incremental-engine: Rejected delta from '${topic}' event - empty or invalid`,
);
}
}
supportsEventTopics(): string[] {
const { provider } = this.options;
const topics = provider.eventHandler
? provider.eventHandler.supportsEventTopics()
: [];
return topics;
}
}
plugins/catalog-backend-module-incremental-ingestion/src/module/WrapperProviders.ts
+3 -3
View File
@@ -31,7 +31,7 @@ import { Duration } from 'luxon';
import { IncrementalIngestionDatabaseManager } from '../database/IncrementalIngestionDatabaseManager';
import { applyDatabaseMigrations } from '../database/migrations';
import { IncrementalIngestionEngine } from '../engine/IncrementalIngestionEngine';
import { createIncrementalProviderRouter } from '../router/routes';
import { IncrementalProviderRouter } from '../router/routes';
import {
IncrementalEntityProvider,
IncrementalEntityProviderOptions,
@@ -75,10 +75,10 @@ export class WrapperProviders {
}
async adminRouter(): Promise<express.Router> {
return createIncrementalProviderRouter(
return await new IncrementalProviderRouter(
new IncrementalIngestionDatabaseManager({ client: this.options.client }),
loggerToWinstonLogger(this.options.logger),
);
).createRouter();
}
private async startProvider(
plugins/catalog-backend-module-incremental-ingestion/src/router/routes.ts
+202 -195
View File
@@ -21,212 +21,219 @@ import { Logger } from 'winston';
import { IncrementalIngestionDatabaseManager } from '../database/IncrementalIngestionDatabaseManager';
import { PROVIDER_BASE_PATH, PROVIDER_CLEANUP, PROVIDER_HEALTH } from './paths';
export const createIncrementalProviderRouter = async (
manager: IncrementalIngestionDatabaseManager,
logger: Logger,
) => {
const router = Router();
router.use(express.json());
export class IncrementalProviderRouter {
private manager: IncrementalIngestionDatabaseManager;
private logger: Logger;
// Get the overall health of all incremental providers
router.get(PROVIDER_HEALTH, async (_, res) => {
const records = await manager.healthcheck();
const providers = records.map(record => record.provider_name);
const duplicates = [
...new Set(providers.filter((e, i, a) => a.indexOf(e) !== i)),
];
constructor(manager: IncrementalIngestionDatabaseManager, logger: Logger) {
this.manager = manager;
this.logger = logger;
}
if (duplicates.length > 0) {
res.json({ healthy: false, duplicateIngestions: duplicates });
} else {
res.json({ healthy: true });
}
});
async createRouter() {
const router = Router();
router.use(express.json());
// Clean up and pause all providers
router.post(PROVIDER_CLEANUP, async (_, res) => {
const result = await manager.cleanupProviders();
res.json(result);
});
// Get the overall health of all incremental providers
router.get(PROVIDER_HEALTH, async (_, res) => {
const records = await this.manager.healthcheck();
const providers = records.map(record => record.provider_name);
const duplicates = [
...new Set(providers.filter((e, i, a) => a.indexOf(e) !== i)),
];
// Get basic status of the provider
router.get(PROVIDER_BASE_PATH, async (req, res) => {
const { provider } = req.params;
const record = await manager.getCurrentIngestionRecord(provider);
if (record) {
res.json({
success: true,
status: {
current_action: record.status,
next_action_at: new Date(record.next_action_at),
},
last_error: record.last_error,
});
} else {
const providers: string[] = await manager.listProviders();
if (providers.includes(provider)) {
if (duplicates.length > 0) {
res.json({ healthy: false, duplicateIngestions: duplicates });
} else {
res.json({ healthy: true });
}
});
// Clean up and pause all providers
router.post(PROVIDER_CLEANUP, async (_, res) => {
const result = await this.manager.cleanupProviders();
res.json(result);
});
// Get basic status of the provider
router.get(PROVIDER_BASE_PATH, async (req, res) => {
const { provider } = req.params;
const record = await this.manager.getCurrentIngestionRecord(provider);
if (record) {
res.json({
success: true,
status: {
current_action: 'rest complete, waiting to start',
current_action: record.status,
next_action_at: new Date(record.next_action_at),
},
last_error: record.last_error,
});
} else {
logger.error(
`${provider} - No ingestion record found in the database!`,
);
res.status(404).json({
success: false,
status: {},
last_error: `Provider '${provider}' not found`,
});
const providers: string[] = await this.manager.listProviders();
if (providers.includes(provider)) {
res.json({
success: true,
status: {
current_action: 'rest complete, waiting to start',
},
});
} else {
this.logger.error(
`${provider} - No ingestion record found in the database!`,
);
res.status(404).json({
success: false,
status: {},
last_error: `Provider '${provider}' not found`,
});
}
}
}
});
// Trigger the provider's next action
router.post(`${PROVIDER_BASE_PATH}/trigger`, async (req, res) => {
const { provider } = req.params;
const record = await manager.getCurrentIngestionRecord(provider);
if (record) {
await manager.triggerNextProviderAction(provider);
res.json({
success: true,
message: `${provider}: Next action triggered.`,
});
} else {
const providers: string[] = await manager.listProviders();
if (providers.includes(provider)) {
logger.debug(`${provider} - Ingestion record found`);
res.json({
success: true,
message: 'Unable to trigger next action (provider is restarting)',
});
} else {
res.status(404).json({
success: false,
message: `Provider '${provider}' not found`,
});
}
}
});
// Start a brand-new ingestion cycle for the provider.
// (Cancel's the current run if active, or marks it complete if resting)
router.post(`${PROVIDER_BASE_PATH}/start`, async (req, res) => {
const { provider } = req.params;
const record = await manager.getCurrentIngestionRecord(provider);
if (record) {
const ingestionId = record.id;
if (record.status === 'resting') {
await manager.setProviderComplete(ingestionId);
} else {
await manager.setProviderCanceling(ingestionId);
}
res.json({
success: true,
message: `${provider}: Next cycle triggered.`,
});
} else {
const providers: string[] = await manager.listProviders();
if (providers.includes(provider)) {
logger.debug(`${provider} - Ingestion record found`);
res.json({
success: true,
message: 'Provider is already restarting',
});
} else {
res.status(404).json({
success: false,
message: `Provider '${provider}' not found`,
});
}
}
});
// Stop the provider and pause it for 24 hours
router.post(`${PROVIDER_BASE_PATH}/cancel`, async (req, res) => {
const { provider } = req.params;
const record = await manager.getCurrentIngestionRecord(provider);
if (record) {
const next_action_at = new Date();
next_action_at.setTime(next_action_at.getTime() + 24 * 60 * 60 * 1000);
await manager.updateByName(provider, {
next_action: 'nothing (done)',
ingestion_completed_at: new Date(),
next_action_at,
status: 'resting',
});
res.json({
success: true,
message: `${provider}: Current ingestion canceled.`,
});
} else {
const providers: string[] = await manager.listProviders();
if (providers.includes(provider)) {
logger.debug(`${provider} - Ingestion record found`);
res.json({
success: true,
message: 'Provider is currently restarting, please wait.',
});
} else {
res.status(404).json({
success: false,
message: `Provider '${provider}' not found`,
});
}
}
});
// Wipe out all ingestion records for the provider and pause for 24 hours
router.delete(PROVIDER_BASE_PATH, async (req, res) => {
const { provider } = req.params;
const result = await manager.purgeAndResetProvider(provider);
res.json(result);
});
// Get the ingestion marks for the current cycle
router.get(`${PROVIDER_BASE_PATH}/marks`, async (req, res) => {
const { provider } = req.params;
const record = await manager.getCurrentIngestionRecord(provider);
if (record) {
const id = record.id;
const records = await manager.getAllMarks(id);
res.json({ success: true, records });
} else {
const providers: string[] = await manager.listProviders();
if (providers.includes(provider)) {
logger.debug(`${provider} - Ingestion record found`);
res.json({
success: true,
message: 'No records yet (provider is restarting)',
});
} else {
logger.error(
`${provider} - No ingestion record found in the database!`,
);
res.status(404).json({
success: false,
status: {},
last_error: `Provider '${provider}' not found`,
});
}
}
});
router.delete(`${PROVIDER_BASE_PATH}/marks`, async (req, res) => {
const { provider } = req.params;
const deletions = await manager.clearFinishedIngestions(provider);
res.json({
success: true,
message: `Expired marks for provider '${provider}' removed.`,
deletions,
});
});
router.use(errorHandler());
// Trigger the provider's next action
router.post(`${PROVIDER_BASE_PATH}/trigger`, async (req, res) => {
const { provider } = req.params;
const record = await this.manager.getCurrentIngestionRecord(provider);
if (record) {
await this.manager.triggerNextProviderAction(provider);
res.json({
success: true,
message: `${provider}: Next action triggered.`,
});
} else {
const providers: string[] = await this.manager.listProviders();
if (providers.includes(provider)) {
this.logger.debug(`${provider} - Ingestion record found`);
res.json({
success: true,
message: 'Unable to trigger next action (provider is restarting)',
});
} else {
res.status(404).json({
success: false,
message: `Provider '${provider}' not found`,
});
}
}
});
return router;
};
// Start a brand-new ingestion cycle for the provider.
// (Cancel's the current run if active, or marks it complete if resting)
router.post(`${PROVIDER_BASE_PATH}/start`, async (req, res) => {
const { provider } = req.params;
const record = await this.manager.getCurrentIngestionRecord(provider);
if (record) {
const ingestionId = record.id;
if (record.status === 'resting') {
await this.manager.setProviderComplete(ingestionId);
} else {
await this.manager.setProviderCanceling(ingestionId);
}
res.json({
success: true,
message: `${provider}: Next cycle triggered.`,
});
} else {
const providers: string[] = await this.manager.listProviders();
if (providers.includes(provider)) {
this.logger.debug(`${provider} - Ingestion record found`);
res.json({
success: true,
message: 'Provider is already restarting',
});
} else {
res.status(404).json({
success: false,
message: `Provider '${provider}' not found`,
});
}
}
});
// Stop the provider and pause it for 24 hours
router.post(`${PROVIDER_BASE_PATH}/cancel`, async (req, res) => {
const { provider } = req.params;
const record = await this.manager.getCurrentIngestionRecord(provider);
if (record) {
const next_action_at = new Date();
next_action_at.setTime(next_action_at.getTime() + 24 * 60 * 60 * 1000);
await this.manager.updateByName(provider, {
next_action: 'nothing (done)',
ingestion_completed_at: new Date(),
next_action_at,
status: 'resting',
});
res.json({
success: true,
message: `${provider}: Current ingestion canceled.`,
});
} else {
const providers: string[] = await this.manager.listProviders();
if (providers.includes(provider)) {
this.logger.debug(`${provider} - Ingestion record found`);
res.json({
success: true,
message: 'Provider is currently restarting, please wait.',
});
} else {
res.status(404).json({
success: false,
message: `Provider '${provider}' not found`,
});
}
}
});
// Wipe out all ingestion records for the provider and pause for 24 hours
router.delete(PROVIDER_BASE_PATH, async (req, res) => {
const { provider } = req.params;
const result = await this.manager.purgeAndResetProvider(provider);
res.json(result);
});
// Get the ingestion marks for the current cycle
router.get(`${PROVIDER_BASE_PATH}/marks`, async (req, res) => {
const { provider } = req.params;
const record = await this.manager.getCurrentIngestionRecord(provider);
if (record) {
const id = record.id;
const records = await this.manager.getAllMarks(id);
res.json({ success: true, records });
} else {
const providers: string[] = await this.manager.listProviders();
if (providers.includes(provider)) {
this.logger.debug(`${provider} - Ingestion record found`);
res.json({
success: true,
message: 'No records yet (provider is restarting)',
});
} else {
this.logger.error(
`${provider} - No ingestion record found in the database!`,
);
res.status(404).json({
success: false,
status: {},
last_error: `Provider '${provider}' not found`,
});
}
}
});
router.delete(`${PROVIDER_BASE_PATH}/marks`, async (req, res) => {
const { provider } = req.params;
const deletions = await this.manager.clearFinishedIngestions(provider);
res.json({
success: true,
message: `Expired marks for provider '${provider}' removed.`,
deletions,
});
});
router.use(errorHandler());
return router;
}
}
plugins/catalog-backend-module-incremental-ingestion/src/service/IncrementalCatalogBuilder.ts
+3 -3
View File
@@ -24,7 +24,7 @@ import { Knex } from 'knex';
import { IncrementalIngestionEngine } from '../engine/IncrementalIngestionEngine';
import { applyDatabaseMigrations } from '../database/migrations';
import { IncrementalIngestionDatabaseManager } from '../database/IncrementalIngestionDatabaseManager';
import { createIncrementalProviderRouter } from '../router/routes';
import { IncrementalProviderRouter } from '../router/routes';
import { Deferred } from '../util';
/** @public */
@@ -60,10 +60,10 @@ export class IncrementalCatalogBuilder {
router: 'IncrementalProviderAdmin',
});
const incrementalAdminRouter = await createIncrementalProviderRouter(
const incrementalAdminRouter = await new IncrementalProviderRouter(
this.manager,
routerLogger,
);
).createRouter();
return { incrementalAdminRouter };
}
plugins/catalog-backend-module-incremental-ingestion/src/types.ts
+32
View File
@@ -27,6 +27,7 @@ import type {
DeferredEntity,
EntityProviderConnection,
} from '@backstage/plugin-catalog-backend';
import { EventParams } from '@backstage/plugin-events-node';
import type { PermissionEvaluator } from '@backstage/plugin-permission-common';
import type { DurationObjectUnits } from 'luxon';
import type { Logger } from 'winston';
@@ -75,6 +76,37 @@ export interface IncrementalEntityProvider<TCursor, TContext> {
* @param burst - a function which performs a series of iterations
*/
around(burst: (context: TContext) => Promise<void>): Promise<void>;
/**
* If set, the IncrementalEntityProvider will receive and respond to
* events.
*
* This system acts as a wrapper for the Backstage events bus, and
* requires the events backend to function. It does not provide its
* own events backend. See {@link https://github.com/backstage/backstage/tree/master/plugins/events-backend}.
*/
eventHandler?: {
/**
* This method accepts an incoming event for the provider, and
* optionally maps the payload to an object containing a delta
* mutation.
*
* If a valid delta is returned by this method, it will be ingested
* automatically by the provider.
*/
onEvent: (params: EventParams) =>
| undefined
| {
added: DeferredEntity[];
removed: { entityRef: string }[];
};
/**
* This method returns an array of topics for the IncrementalEntityProvider
* to respond to.
*/
supportsEventTopics: () => string[];
};
}
/**
yarn.lock
+1
View File
@@ -4989,6 +4989,7 @@ __metadata:
"@backstage/errors": "workspace:^"
"@backstage/plugin-catalog-backend": "workspace:^"
"@backstage/plugin-catalog-node": "workspace:^"
"@backstage/plugin-events-node": "workspace:^"
"@backstage/plugin-permission-common": "workspace:^"
"@types/express": ^4.17.6
"@types/luxon": ^3.0.0