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

shave down the api report, some final fixes

Browse Source 
Signed-off-by: Fredrik Adelöw <freben@gmail.com>
This commit is contained in:
Fredrik Adelöw
2022-11-22 11:30:35 +01:00
parent 6404ad16aa
commit 35455be414
13 changed files with 113 additions and 271 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/loud-rockets-reply.md
+2 -2
View File
@@ -1,5 +1,5 @@
---
'@backstage/plugin-incremental-ingestion-backend': patch
'@backstage/plugin-incremental-ingestion-backend': minor
---
This change introduces incremental entity providers, which are used for streaming very large data sources into the catalog.
Introduces incremental entity providers, which are used for streaming very large data sources into the catalog.
plugins/incremental-ingestion-backend/README.md
+74 -71
View File
@@ -1,4 +1,4 @@
# backend-incremental-ingestion
# `@backstage/plugin-incremental-ingestion-backend`
The Incremental Ingestion Backend plugin provides an Incremental Entity Provider that can be used to ingest data from sources using delta mutations, while retaining the orphan prevention mechanism provided by full mutations.
@@ -45,63 +45,66 @@ The Incremental Entity Provider backend is designed for data sources that provid
1. Install `@backstage/plugin-incremental-ingestion-backend` with `yarn workspace backend add @backstage/plugin-incremental-ingestion-backend`
2. Import `IncrementalCatalogBuilder` from `@backstage/plugin-incremental-ingestion-backend` and instantiate it with `await IncrementalCatalogBuilder.create(env, builder)`. You have to pass `builder` into `IncrementalCatalogBuilder.create` function because `IncrementalCatalogBuilder` will convert an `IncrementalEntityProvider` into an `EntityProvider` and call `builder.addEntityProvider`.
```ts
const builder = CatalogBuilder.create(env);
// incremental builder receives builder because it'll register
// incremental entity providers with the builder
const incrementalBuilder = await IncrementalCatalogBuilder.create(env, builder);
```
```ts
const builder = CatalogBuilder.create(env);
// incremental builder receives builder because it'll register
// incremental entity providers with the builder
const incrementalBuilder = await IncrementalCatalogBuilder.create(
env,
builder,
);
```
3. Last step, add `await incrementBuilder.build()` after `await builder.build()` to ensure that all `CatalogBuider` migration run before running `incrementBuilder.build()` migrations.
3. Last step, add `await incrementBuilder.build()` after `await builder.build()` to ensure that all `CatalogBuilder` migration run before running `incrementBuilder.build()` migrations.
```ts
const { processingEngine, router } = await builder.build();
```ts
const { processingEngine, router } = await builder.build();
// this has to run after `await builder.build()` to ensure that catalog migrations are completed
// before incremental builder migrations are executed
await incrementalBuilder.build();
```
// this has to run after `await builder.build()` to ensure that catalog migrations are completed
// before incremental builder migrations are executed
await incrementalBuilder.build();
```
The result should look something like this,
The result should look something like this,
```ts
import { CatalogBuilder } from '@backstage/plugin-catalog-backend';
import { ScaffolderEntitiesProcessor } from '@backstage/plugin-scaffolder-backend';
import { IncrementalCatalogBuilder } from '@backstage/plugin-incremental-ingestion-backend';
import { Router } from 'express';
import { Duration } from 'luxon';
import { PluginEnvironment } from '../types';
```ts
import { CatalogBuilder } from '@backstage/plugin-catalog-backend';
import { ScaffolderEntitiesProcessor } from '@backstage/plugin-scaffolder-backend';
import { IncrementalCatalogBuilder } from '@backstage/plugin-incremental-ingestion-backend';
import { Router } from 'express';
import { Duration } from 'luxon';
import { PluginEnvironment } from '../types';
export default async function createPlugin(
env: PluginEnvironment,
): Promise<Router> {
const builder = CatalogBuilder.create(env);
// incremental builder receives builder because it'll register
// incremental entity providers with the builder
const incrementalBuilder = await IncrementalCatalogBuilder.create(
env,
builder,
);
export default async function createPlugin(
env: PluginEnvironment,
): Promise<Router> {
const builder = CatalogBuilder.create(env);
// incremental builder receives builder because it'll register
// incremental entity providers with the builder
const incrementalBuilder = await IncrementalCatalogBuilder.create(
env,
builder,
);
builder.addProcessor(new ScaffolderEntitiesProcessor());
builder.addProcessor(new ScaffolderEntitiesProcessor());
const { processingEngine, router } = await builder.build();
const { processingEngine, router } = await builder.build();
// this has to run after `await builder.build()` so ensure that catalog migrations are completed
// before incremental builder migrations are executed
const { incrementalAdminRouter } = await incrementalBuilder.build();
// this has to run after `await builder.build()` so ensure that catalog migrations are completed
// before incremental builder migrations are executed
const { incrementalAdminRouter } = await incrementalBuilder.build();
router.use('/incremental', incrementalAdminRouter);
router.use('/incremental', incrementalAdminRouter);
await processingEngine.start();
await processingEngine.start();
return router;
}
```
return router;
}
```
## Writing an Incremental Entity Provider
To create an Incremental Entity Provider, you need to know how to retrieve a single page of the data that you wish to ingest into the Backstage catalog. If the API has pagination and you know how to make a paginated request to that API, you'll be able to implement an Incremental Entity Provider for this API. For more information about compatibility, checkout <a href="#Compatible-Data-Source">Compatible data sources</a> section on this page.
To create an Incremental Entity Provider, you need to know how to retrieve a single page of the data that you wish to ingest into the Backstage catalog. If the API has pagination and you know how to make a paginated request to that API, you'll be able to implement an Incremental Entity Provider for this API. For more information about compatibility, checkout <a href="#compatible-data-source">Compatible data sources</a> section on this page.
Here is the type definition for an Incremental Entity Provider.
@@ -127,7 +130,7 @@ interface IncrementalEntityProvider<TCursor, TContext> {
* ingestion.
*
* @param context - anything needed in order to fetch a single page.
* @param cursor - a uniqiue value identifying the page to ingest.
* @param cursor - a unique value identifying the page to ingest.
* @returns the entities to be ingested, as well as the cursor of
* the the next page after this one.
*/
@@ -230,7 +233,7 @@ export class MyIncrementalEntityProvider
The last step is to implement the actual `next` method that will accept the cursor, call the API, process the result and return the result.
```ts
export class MyIncrementalEntityProvider implements IncrementalEntityProvider<Cursor, Context> {
export class MyIncrementalEntityProvider implements IncrementalEntityProvider<MyApiCursor, MyContext> {
token: string;
@@ -250,7 +253,7 @@ export class MyIncrementalEntityProvider implements IncrementalEntityProvider<Cu
await burst({ apiClient })
}
async next(context: MyContext, cursor?: MyApiCursor = { page: 1 }): Promise<EntityIteratorResult<TCursor>> {
async next(context: MyContext, cursor?: MyApiCursor = { page: 1 }): Promise<EntityIteratorResult<MyApiCursor>> {
const { apiClient } = context;
// call your API with the current cursor
@@ -299,36 +302,36 @@ Now that you have your new Incremental Entity Provider, we can connect it to the
## Adding an Incremental Entity Provider to the catalog
We'll assume you followed the <a href="#Installation">Installation</a> instructions. After you create your `incrementalBuilder`, you can instantiate your Entity Provider and pass it to the `addIncrementalEntityProvider` method.
We'll assume you followed the <a href="#installation">Installation</a> instructions. After you create your `incrementalBuilder`, you can instantiate your Entity Provider and pass it to the `addIncrementalEntityProvider` method.
```ts
const incrementalBuilder = await IncrementalCatalogBuilder.create(env, builder);
const incrementalBuilder = await IncrementalCatalogBuilder.create(env, builder);
// I'm assuming you're going to get your token from config
const token = config.getString('myApiClient.token');
// I'm assuming you're going to get your token from config
const token = config.getString('myApiClient.token');
const myEntityProvider = new MyIncrementalEntityProvider(token)
const myEntityProvider = new MyIncrementalEntityProvider(token)
incrementalBuilder.addIncrementalEntityProvider(
myEntityProvider,
{
// how long should it attempt to read pages from the API
// keep this short. Incremental Entity Provider will attempt to
// read as many pages as it can in this time
burstLength: Duration.fromObject({ seconds: 3 }),
// how long should it wait between bursts?
burstInterval: Duration.fromObject({ seconds: 3 }),
// how long should it rest before re-ingesting again?
restLength: Duration.fromObject({ day: 1 })
// optional back-off configuration - how long should it wait to retry?
backoff: [
Duration.fromObject({ seconds: 5 }),
Duration.fromObject({ seconds: 30 }),
Duration.fromObject({ minutes: 10 }),
Duration.fromObject({ hours: 3 })
]
}
)
incrementalBuilder.addIncrementalEntityProvider(
myEntityProvider,
{
// how long should it attempt to read pages from the API
// keep this short. Incremental Entity Provider will attempt to
// read as many pages as it can in this time
burstLength: Duration.fromObject({ seconds: 3 }),
// how long should it wait between bursts?
burstInterval: Duration.fromObject({ seconds: 3 }),
// how long should it rest before re-ingesting again?
restLength: Duration.fromObject({ day: 1 })
// optional back-off configuration - how long should it wait to retry?
backoff: [
Duration.fromObject({ seconds: 5 }),
Duration.fromObject({ seconds: 30 }),
Duration.fromObject({ minutes: 10 }),
Duration.fromObject({ hours: 3 })
]
}
)
```
That's it!!!
plugins/incremental-ingestion-backend/api-report.md
+4 -165
View File
@@ -8,11 +8,9 @@
import { CatalogBuilder } from '@backstage/plugin-catalog-backend';
import type { Config } from '@backstage/config';
import type { DeferredEntity } from '@backstage/plugin-catalog-backend';
import { Duration } from 'luxon';
import type { DurationObjectUnits } from 'luxon';
import { Knex } from 'knex';
import type { Logger } from 'winston';
import type { PermissionAuthorizer } from '@backstage/plugin-permission-common';
import type { PermissionEvaluator } from '@backstage/plugin-permission-common';
import type { PluginDatabaseManager } from '@backstage/backend-common';
import type { PluginTaskScheduler } from '@backstage/backend-tasks';
import { Router } from 'express';
@@ -38,14 +36,13 @@ export const INCREMENTAL_ENTITY_PROVIDER_ANNOTATION =
// @public (undocumented)
export class IncrementalCatalogBuilder {
// (undocumented)
addIncrementalEntityProvider<T, C>(
provider: IncrementalEntityProvider<T, C>,
addIncrementalEntityProvider<TCursor, TContext>(
provider: IncrementalEntityProvider<TCursor, TContext>,
options: IncrementalEntityProviderOptions,
): void;
// (undocumented)
build(): Promise<{
incrementalAdminRouter: Router;
manager: IncrementalIngestionDatabaseManager;
}>;
static create(
env: PluginEnvironment,
@@ -71,162 +68,6 @@ export interface IncrementalEntityProviderOptions {
restLength: DurationObjectUnits;
}
// @public (undocumented)
export class IncrementalIngestionDatabaseManager {
constructor(options: { client: Knex });
cleanupProviders(): Promise<{
ingestionsDeleted: void;
ingestionMarksDeleted: void;
markEntitiesDeleted: void;
}>;
clearDuplicateIngestions(
ingestionId: string,
provider: string,
): Promise<void>;
clearFinishedIngestions(provider: string): Promise<{
deletions: {
markEntitiesDeleted: number;
marksDeleted: number;
ingestionsDeleted: number;
};
}>;
computeRemoved(
provider: string,
ingestionId: string,
): Promise<
{
entity: any;
}[]
>;
createMark(options: MarkRecordInsert): Promise<void>;
createMarkEntities(markId: string, entities: DeferredEntity[]): Promise<void>;
createProviderIngestionRecord(provider: string): Promise<
| {
ingestionId: string;
nextAction: string;
attempts: number;
nextActionAt: number;
}
| undefined
>;
// (undocumented)
getAllMarks(ingestionId: string): Promise<MarkRecord[]>;
getCurrentIngestionRecord(
provider: string,
): Promise<IngestionRecord | undefined>;
getLastMark(ingestionId: string): Promise<MarkRecord | undefined>;
healthcheck(): Promise<
Pick<
{
id: string;
provider_name: string;
},
'id' | 'provider_name'
>[]
>;
insertIngestionRecord(record: IngestionUpsertIFace): Promise<void>;
listProviders(): Promise<string[]>;
purgeAndResetProvider(provider: string): Promise<{
provider: string;
ingestionsDeleted: number;
marksDeleted: number;
markEntitiesDeleted: number;
}>;
purgeTable(table: string): Promise<void>;
setProviderBackoff(
ingestionId: string,
attempts: number,
error: Error,
backoffLength: number,
): Promise<void>;
setProviderBursting(ingestionId: string): Promise<void>;
setProviderCanceled(ingestionId: string): Promise<void>;
setProviderCanceling(ingestionId: string, message?: string): Promise<void>;
setProviderComplete(ingestionId: string): Promise<void>;
setProviderIngesting(ingestionId: string): Promise<void>;
setProviderInterstitial(ingestionId: string): Promise<void>;
setProviderResting(ingestionId: string, restLength: Duration): Promise<void>;
triggerNextProviderAction(provider: string): Promise<void>;
// (undocumented)
updateByName(
provider: string,
update: Partial<IngestionUpsertIFace>,
): Promise<void>;
updateIngestionRecordById(options: IngestionRecordUpdate): Promise<void>;
updateIngestionRecordByProvider(
provider: string,
update: Partial<IngestionUpsertIFace>,
): Promise<void>;
}
// @public
export interface IngestionRecord extends IngestionUpsertIFace {
created_at: string;
// (undocumented)
id: string;
// (undocumented)
next_action_at: Date;
}
// @public
export interface IngestionRecordUpdate {
// (undocumented)
ingestionId: string;
// (undocumented)
update: Partial<IngestionUpsertIFace>;
}
// @public
export interface IngestionUpsertIFace {
attempts?: number;
completion_ticket: string;
id?: string;
ingestion_completed_at?: Date | string | null;
last_error?: string | null;
next_action:
| 'rest'
| 'ingest'
| 'backoff'
| 'cancel'
| 'nothing (done)'
| 'nothing (canceled)';
next_action_at?: Date;
provider_name: string;
rest_completed_at?: Date | string | null;
status:
| 'complete'
| 'bursting'
| 'resting'
| 'canceling'
| 'interstitial'
| 'backing off';
}
// @public
export interface MarkRecord {
// (undocumented)
created_at: string;
// (undocumented)
cursor: string;
// (undocumented)
id: string;
// (undocumented)
ingestion_id: string;
// (undocumented)
sequence: number;
}
// @public
export interface MarkRecordInsert {
// (undocumented)
record: {
id: string;
ingestion_id: string;
cursor: unknown;
sequence: number;
};
}
// @public (undocumented)
export type PluginEnvironment = {
logger: Logger;
@@ -234,8 +75,6 @@ export type PluginEnvironment = {
scheduler: PluginTaskScheduler;
config: Config;
reader: UrlReader;
permissions: PermissionAuthorizer;
permissions: PermissionEvaluator;
};
// (No @packageDocumentation comment for this package)
```
plugins/incremental-ingestion-backend/migrations/20221116073152_init.js
+1 -1
View File
@@ -54,7 +54,7 @@ exports.up = async function up(knex) {
table
.string('last_error')
.comment('records any error that occured in the previous burst attempt');
.comment('records any error that occurred in the previous burst attempt');
table
.integer('attempts')
plugins/incremental-ingestion-backend/package.json
+12 -1
View File
@@ -1,7 +1,7 @@
{
"name": "@backstage/plugin-incremental-ingestion-backend",
"description": "An entity provider for streaming large asset sources into the catalog",
"version": "0.1.0",
"version": "0.0.0",
"main": "src/index.ts",
"types": "src/index.ts",
"license": "Apache-2.0",
@@ -13,6 +13,15 @@
"backstage": {
"role": "backend-plugin"
},
"homepage": "https://backstage.io",
"repository": {
"type": "git",
"url": "https://github.com/backstage/backstage",
"directory": "plugins/incremental-ingestion-backend"
},
"keywords": [
"backstage"
],
"scripts": {
"start": "backstage-cli package start",
"build": "backstage-cli package build",
@@ -24,10 +33,12 @@
},
"dependencies": {
"@backstage/backend-common": "workspace:^",
"@backstage/backend-plugin-api": "workspace:^",
"@backstage/backend-tasks": "workspace:^",
"@backstage/catalog-model": "workspace:^",
"@backstage/config": "workspace:^",
"@backstage/plugin-catalog-backend": "workspace:^",
"@backstage/plugin-catalog-node": "workspace:^",
"@backstage/plugin-permission-common": "workspace:^",
"@types/express": "^4.17.6",
"@types/luxon": "^3.0.0",
plugins/incremental-ingestion-backend/src/database/IncrementalIngestionDatabaseManager.ts
-1
View File
@@ -28,7 +28,6 @@ import {
MarkRecordInsert,
} from '../types';
/** @public */
export class IncrementalIngestionDatabaseManager {
private client: Knex;
plugins/incremental-ingestion-backend/src/database/tables.ts
+1 -2
View File
@@ -14,5 +14,4 @@
* limitations under the License.
*/
export const DB_MIGRATIONS_TABLE =
'plugin_incremental_ingestion_backend__knex_migrations';
export const DB_MIGRATIONS_TABLE = 'incremental_ingestion__knex_migrations';
plugins/incremental-ingestion-backend/src/engine/IncrementalIngestionEngine.ts
-1
View File
@@ -27,7 +27,6 @@ import { performance } from 'perf_hooks';
import { Duration, DurationObjectUnits } from 'luxon';
import { v4 } from 'uuid';
/** @public */
export class IncrementalIngestionEngine implements IterationEngine {
restLength: Duration;
backoff: DurationObjectUnits[];
plugins/incremental-ingestion-backend/src/index.ts
+7 -6
View File
@@ -13,17 +13,18 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
/**
* Provides efficient incremental ingestion of entities into the catalog.
*
* @packageDocumentation
*/
export { IncrementalCatalogBuilder } from './service/IncrementalCatalogBuilder';
export type { IncrementalIngestionDatabaseManager } from './database/IncrementalIngestionDatabaseManager';
export type {
EntityIteratorResult,
IncrementalEntityProvider,
IncrementalEntityProviderOptions,
IngestionRecord,
IngestionRecordUpdate,
IngestionUpsert as IngestionUpsertIFace,
MarkRecord,
MarkRecordInsert,
INCREMENTAL_ENTITY_PROVIDER_ANNOTATION,
PluginEnvironment,
} from './types';
plugins/incremental-ingestion-backend/src/router/routes.ts
+2 -2
View File
@@ -13,14 +13,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import { errorHandler } from '@backstage/backend-common';
import express from 'express';
import Router from 'express-promise-router';
import { Logger } from 'winston';
import { IncrementalIngestionDatabaseManager } from '..';
import { IncrementalIngestionDatabaseManager } from '../database/IncrementalIngestionDatabaseManager';
import { PROVIDER_BASE_PATH, PROVIDER_CLEANUP, PROVIDER_HEALTH } from './paths';
/** @public */
export const createIncrementalProviderRouter = async (
manager: IncrementalIngestionDatabaseManager,
logger: Logger,
plugins/incremental-ingestion-backend/src/service/IncrementalCatalogBuilder.ts
+3 -3
View File
@@ -93,11 +93,11 @@ export class IncrementalCatalogBuilder {
routerLogger,
);
return { incrementalAdminRouter, manager: this.manager };
return { incrementalAdminRouter };
}
addIncrementalEntityProvider<T, C>(
provider: IncrementalEntityProvider<T, C>,
addIncrementalEntityProvider<TCursor, TContext>(
provider: IncrementalEntityProvider<TCursor, TContext>,
options: IncrementalEntityProviderOptions,
) {
const { burstInterval, burstLength, restLength } = options;
plugins/incremental-ingestion-backend/src/types.ts
+5 -16
View File
@@ -26,10 +26,11 @@ import type {
DeferredEntity,
EntityProviderConnection,
} from '@backstage/plugin-catalog-backend';
import type { PermissionAuthorizer } from '@backstage/plugin-permission-common';
import type { PermissionEvaluator } from '@backstage/plugin-permission-common';
import type { DurationObjectUnits } from 'luxon';
import type { Logger } from 'winston';
import { IncrementalIngestionDatabaseManager } from './';
import { IncrementalIngestionDatabaseManager } from './database/IncrementalIngestionDatabaseManager';
/**
* Entity annotation containing the incremental entity provider.
*
@@ -64,7 +65,7 @@ export interface IncrementalEntityProvider<TCursor, TContext> {
* ingestion.
*
* @param context - anything needed in order to fetch a single page.
* @param cursor - a uniqiue value identifying the page to ingest.
* @param cursor - a unique value identifying the page to ingest.
* @returns The entities to be ingested, as well as the cursor of
* the next page after this one.
*/
@@ -137,15 +138,13 @@ export type PluginEnvironment = {
scheduler: PluginTaskScheduler;
config: Config;
reader: UrlReader;
permissions: PermissionAuthorizer;
permissions: PermissionEvaluator;
};
/** @public */
export interface IterationEngine {
taskFn: TaskFunction;
}
/** @public */
export interface IterationEngineOptions {
logger: Logger;
connection: EntityProviderConnection;
@@ -158,8 +157,6 @@ export interface IterationEngineOptions {
/**
* The shape of data inserted into or updated in the `ingestions` table.
*
* @public
*/
export interface IngestionUpsert {
/**
@@ -218,8 +215,6 @@ export interface IngestionUpsert {
/**
* This interface is for updating an existing ingestion record.
*
* @public
*/
export interface IngestionRecordUpdate {
ingestionId: string;
@@ -228,8 +223,6 @@ export interface IngestionRecordUpdate {
/**
* The expected response from the `ingestion_marks` table.
*
* @public
*/
export interface MarkRecord {
id: string;
@@ -241,8 +234,6 @@ export interface MarkRecord {
/**
* The expected response from the `ingestions` table.
*
* @public
*/
export interface IngestionRecord extends IngestionUpsert {
id: string;
@@ -255,8 +246,6 @@ export interface IngestionRecord extends IngestionUpsert {
/**
* This interface supplies all the values for adding an ingestion mark.
*
* @public
*/
export interface MarkRecordInsert {
record: {
yarn.lock
+2
View File
@@ -6028,11 +6028,13 @@ __metadata:
resolution: "@backstage/plugin-incremental-ingestion-backend@workspace:plugins/incremental-ingestion-backend"
dependencies:
"@backstage/backend-common": "workspace:^"
"@backstage/backend-plugin-api": "workspace:^"
"@backstage/backend-tasks": "workspace:^"
"@backstage/catalog-model": "workspace:^"
"@backstage/cli": "workspace:^"
"@backstage/config": "workspace:^"
"@backstage/plugin-catalog-backend": "workspace:^"
"@backstage/plugin-catalog-node": "workspace:^"
"@backstage/plugin-permission-common": "workspace:^"
"@types/express": ^4.17.6
"@types/luxon": ^3.0.0