Merge pull request #14356 from dekoding/dekoding/incremental-entity-provider
Adding the incremental entity provider backend
This commit is contained in:
Fredrik Adelöw
@@ -0,0 +1 @@
|
||||
module.exports = require('@backstage/cli/config/eslint-factory')(__dirname);
|
||||
@@ -0,0 +1,341 @@
|
||||
# `@backstage/plugin-catalog-backend-module-incremental-ingestion`
|
||||
|
||||
The Incremental Ingestion catalog backend module 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.
|
||||
|
||||
## Why did we create it?
|
||||
|
||||
Backstage provides an [Entity Provider mechanism that has two kinds of mutations](https://backstage.io/docs/features/software-catalog/external-integrations#provider-mutations): `delta` and `full`. `delta` mutations tell Backstage Catalog which entities should be added and removed from the catalog. `full` mutation accepts a list of entities and automatically computes which entities must be removed by comparing the provided entities against existing entities to create a diff between the two sets. These two kinds of mutations are convenient for different kinds of data sources. A `delta` mutation can be used with a data source that emits UPDATE and DELETE events for its data. A `full` mutation is useful for APIs that produce fewer entities than can fit in Backstage processes' memory.
|
||||
|
||||
Unfortunately, these two kinds of mutations are insufficient for very large data sources for the following reasons,
|
||||
|
||||
1. Even when the API provides DELETE events, we still need a way to create the initial list of entities. For example, if you ingest all repositories from GitHub into Backstage and you use webhooks, you still need the initial list of entities.
|
||||
2. A `delta` mutation can not guarantee that mutations will not be missed. For example, if your Backstage portal is down while receiving a DELETE event, you might miss the event which leaves your catalog in an unclear state. How can you replay the missed events? Some data sources, like GitHub, provide an API for replaying missed events, but this increases complexity and is not available on all APIs.
|
||||
3. Addressing the above two use case with `full` mutation is not an option on very large datasets because a `full` mutation requires that all entities are in memory to create a diff. If your data source has 100k+ records, this can easily cause your processes to run out of memory.
|
||||
4. In cases when you can use `full` mutation, committing many entities into the processing pipeline fills up the processing queue and delays the processing of entities from other entity providers.
|
||||
|
||||
We created the Incremental Entity Provider to address all of the above issues. The Incremental Entity Provider addresses these issues with a combination of `delta` mutations and a mark-and-sweep mechanism. Instead of doing a single `full` mutation, it performs a series of bursts. At the end of each burst, the Incremental Entity Provider performs the following three operations,
|
||||
|
||||
1. Marks each received entity in the database.
|
||||
2. Annotates each entity with `backstage/incremental-entity-provider: <entity-provider-id>` annotation.
|
||||
3. Commits all of the entities with a `delta` mutation.
|
||||
|
||||
Incremental Entity Providers will wait a configurable interval before proceeding to the next burst.
|
||||
|
||||
Once the source has no more results, Incremental Entity Provider compares all entities annotated with `@backstage/incremental-entity-provider: <entity-provider-id>` against all marked entities to determine which entities committed by same entity provider were not marked during the last ingestion cycle. All unmarked entities are deleted at the end of the cycle. The Incremental Entity Provider rests for a fixed internal before restarting the ingestion process.
|
||||
|
||||
|
||||
|
||||
This approach has the following benefits,
|
||||
|
||||
1. Reduced ingestion latency - each burst commits entities which are processed before the entire list is processed.
|
||||
2. Stable pressure - each period between bursts provides an opportunity for the processing pipeline to settle without overwhelming the pipeline with a large number of unprocessed entities.
|
||||
3. Built-in retry / back-off - Failed bursts are automatically retried with a built-in back-off interval providing an opportunity for the data source to reset its rate limits before retrying the burst.
|
||||
4. Prevents orphan entities - Deleted entities are removed as with `full` mutation with a low memory footprint.
|
||||
|
||||
## Requirements
|
||||
|
||||
The Incremental Entity Provider backend is designed for data sources that provide paginated results. Each burst attempts to handle one or more pages of the query. The plugin will attempt to fetch as many pages as it can within a configurable burst length. At every iteration, it expects to receive the next cursor that will be used to query in the next iteration. Each iteration may happen on a different replica. This has several consequences:
|
||||
|
||||
1. The cursor must be serializable to JSON (not an issue for most RESTful or GraphQL based APIs).
|
||||
2. The client must be stateless - a client is created from scratch for each iteration to allow distributing processing over multiple replicas.
|
||||
3. There must be sufficient storage in Postgres to handle the additional data. (Presumably, this is also true of sqlite, but it has only been tested with Postgres.)
|
||||
|
||||
## Installation
|
||||
|
||||
1. Install `@backstage/plugin-catalog-backend-module-incremental-ingestion` with `yarn workspace backend add @backstage/plugin-catalog-backend-module-incremental-ingestion`
|
||||
2. Import `IncrementalCatalogBuilder` from `@backstage/plugin-catalog-backend-module-incremental-ingestion` 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,
|
||||
);
|
||||
```
|
||||
|
||||
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();
|
||||
|
||||
// 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,
|
||||
|
||||
```ts
|
||||
import { CatalogBuilder } from '@backstage/plugin-catalog-backend';
|
||||
import { ScaffolderEntitiesProcessor } from '@backstage/plugin-scaffolder-backend';
|
||||
import { IncrementalCatalogBuilder } from '@backstage/plugin-catalog-backend-module-incremental-ingestion';
|
||||
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,
|
||||
);
|
||||
|
||||
builder.addProcessor(new ScaffolderEntitiesProcessor());
|
||||
|
||||
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();
|
||||
|
||||
router.use('/incremental', incrementalAdminRouter);
|
||||
|
||||
await processingEngine.start();
|
||||
|
||||
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.
|
||||
|
||||
Here is the type definition for an Incremental Entity Provider.
|
||||
|
||||
```ts
|
||||
interface IncrementalEntityProvider<TCursor, TContext> {
|
||||
/**
|
||||
* This name must be unique between all of the entity providers
|
||||
* operating in the catalog.
|
||||
*/
|
||||
getProviderName(): string;
|
||||
|
||||
/**
|
||||
* Do any setup and teardown necessary in order to provide the
|
||||
* context for fetching pages. This should always invoke `burst` in
|
||||
* order to fetch the individual pages.
|
||||
*
|
||||
* @param burst - a function which performs a series of iterations
|
||||
*/
|
||||
around(burst: (context: TContext) => Promise<void>): Promise<void>;
|
||||
|
||||
/**
|
||||
* Return a single page of entities from a specific point in the
|
||||
* ingestion.
|
||||
*
|
||||
* @param context - anything needed in order to fetch a single page.
|
||||
* @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.
|
||||
*/
|
||||
next(
|
||||
context: TContext,
|
||||
cursor?: TCursor,
|
||||
): Promise<EntityIteratorResult<TCursor>>;
|
||||
}
|
||||
```
|
||||
|
||||
For tutorial, we'll write an Incremental Entity Provider that will call an imaginary API. This imaginary API will return a list of imaginary services. This imaginary API has an imaginary API client with the following interface.
|
||||
|
||||
```ts
|
||||
interface MyApiClient {
|
||||
getServices(page: number): MyPaginatedResults<Service>;
|
||||
}
|
||||
|
||||
interface MyPaginatedResults<T> {
|
||||
items: T[];
|
||||
totalPages: number;
|
||||
}
|
||||
|
||||
interface Service {
|
||||
name: string;
|
||||
}
|
||||
```
|
||||
|
||||
These are the only 3 methods that you need to implement. `getProviderName()` is pretty self explanatory and it's exactly same as on Entity Provider.
|
||||
|
||||
```ts
|
||||
import {
|
||||
IncrementalEntityProvider,
|
||||
EntityIteratorResult,
|
||||
} from '@backstage/plugin-catalog-backend-module-incremental-ingestion';
|
||||
|
||||
// this will include your pagination information, let's say our API accepts a `page` parameter.
|
||||
// In this case, the cursor will include `page`
|
||||
interface MyApiCursor {
|
||||
page: number;
|
||||
}
|
||||
|
||||
// This interface describes the type of data that will be passed to your burst function.
|
||||
interface MyContext {
|
||||
apiClient: MyApiClient;
|
||||
}
|
||||
|
||||
export class MyIncrementalEntityProvider
|
||||
implements IncrementalEntityProvider<MyApiCursor, MyContext>
|
||||
{
|
||||
getProviderName() {
|
||||
return `MyIncrementalEntityProvider`;
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
`around` method is used for setup and tear-down. For example, if you need to create a client that will connect to the API, you would do that here.
|
||||
|
||||
```ts
|
||||
export class MyIncrementalEntityProvider
|
||||
implements IncrementalEntityProvider<Cursor, Context>
|
||||
{
|
||||
getProviderName() {
|
||||
return `MyIncrementalEntityProvider`;
|
||||
}
|
||||
|
||||
async around(burst: (context: MyContext) => Promise<void>): Promise<void> {
|
||||
const apiClient = new MyApiClient();
|
||||
|
||||
await burst({ apiClient });
|
||||
|
||||
// if you need to do any teardown, you can do it here
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
If you need to pass a token to your API, then you can create a constructor that will receive a token and use the token to setup the client.
|
||||
|
||||
```ts
|
||||
export class MyIncrementalEntityProvider
|
||||
implements IncrementalEntityProvider<Cursor, Context>
|
||||
{
|
||||
token: string;
|
||||
|
||||
constructor(token: string) {
|
||||
this.token = token;
|
||||
}
|
||||
|
||||
getProviderName() {
|
||||
return `MyIncrementalEntityProvider`;
|
||||
}
|
||||
|
||||
async around(burst: (context: MyContext) => Promise<void>): Promise<void> {
|
||||
const apiClient = new MyApiClient(this.token);
|
||||
|
||||
await burst({ apiClient });
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
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<MyApiCursor, MyContext> {
|
||||
|
||||
token: string;
|
||||
|
||||
constructor(token: string) {
|
||||
this.token = token;
|
||||
}
|
||||
|
||||
getProviderName() {
|
||||
return `MyIncrementalEntityProvider`;
|
||||
}
|
||||
|
||||
|
||||
async around(burst: (context: MyContext) => Promise<void>): Promise<void> {
|
||||
|
||||
const apiClient = new MyApiClient(this.token)
|
||||
|
||||
await burst({ apiClient })
|
||||
}
|
||||
|
||||
async next(context: MyContext, cursor?: MyApiCursor = { page: 1 }): Promise<EntityIteratorResult<MyApiCursor>> {
|
||||
const { apiClient } = context;
|
||||
|
||||
// call your API with the current cursor
|
||||
const data = await apiClient.getServices(cursor);
|
||||
|
||||
// calculate the next page
|
||||
const nextPage = page + 1;
|
||||
|
||||
// figure out if there are any more pages to fetch
|
||||
const done = nextPage > data.totalPages;
|
||||
|
||||
// convert returned items into entities
|
||||
const entities = data.items.map(item => ({
|
||||
entity: {
|
||||
apiVersion: 'backstage.io/v1beta1',
|
||||
kind: 'Component',
|
||||
metadata: {
|
||||
name: item.name,
|
||||
annotations: {
|
||||
// You need to define these, otherwise they'll fail validation
|
||||
[ANNOTATION_LOCATION]: this.getProviderName(),
|
||||
[ANNOTATION_ORIGIN_LOCATION]: this.getProviderName(),
|
||||
}
|
||||
}
|
||||
spec: {
|
||||
type: 'service'
|
||||
}
|
||||
}
|
||||
}));
|
||||
|
||||
// create the next cursor
|
||||
const nextCursor = {
|
||||
page: nextPage
|
||||
};
|
||||
|
||||
return {
|
||||
done,
|
||||
entities,
|
||||
cursor: nextCursor
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Now that you have your new Incremental Entity Provider, we can connect it to the catalog.
|
||||
|
||||
## 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.
|
||||
|
||||
```ts
|
||||
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');
|
||||
|
||||
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 })
|
||||
]
|
||||
}
|
||||
)
|
||||
```
|
||||
|
||||
That's it!!!
|
||||
|
||||
## Error handling
|
||||
|
||||
If `around` or `next` methods throw an error, the error will show up in logs and it'll trigger the Incremental Entity Provider to try again after a back-off period. It'll keep trying until it reaches the last back-off attempt, at which point it will cancel the current ingestion and start over. You don't need to do anything special to handle the retry logic.
|
||||
@@ -0,0 +1,89 @@
|
||||
## API Report File for "@backstage/plugin-catalog-backend-module-incremental-ingestion"
|
||||
|
||||
> Do not edit this file. It is a report generated by [API Extractor](https://api-extractor.com/).
|
||||
|
||||
```ts
|
||||
/// <reference types="express" />
|
||||
|
||||
import { BackendFeature } from '@backstage/backend-plugin-api';
|
||||
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 type { Logger } from 'winston';
|
||||
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';
|
||||
import type { UrlReader } from '@backstage/backend-common';
|
||||
|
||||
// @public
|
||||
export type EntityIteratorResult<T> =
|
||||
| {
|
||||
done: false;
|
||||
entities: DeferredEntity[];
|
||||
cursor: T;
|
||||
}
|
||||
| {
|
||||
done: true;
|
||||
entities?: DeferredEntity[];
|
||||
cursor?: T;
|
||||
};
|
||||
|
||||
// @public
|
||||
export const INCREMENTAL_ENTITY_PROVIDER_ANNOTATION =
|
||||
'backstage.io/incremental-provider-name';
|
||||
|
||||
// @public (undocumented)
|
||||
export class IncrementalCatalogBuilder {
|
||||
// (undocumented)
|
||||
addIncrementalEntityProvider<TCursor, TContext>(
|
||||
provider: IncrementalEntityProvider<TCursor, TContext>,
|
||||
options: IncrementalEntityProviderOptions,
|
||||
): void;
|
||||
// (undocumented)
|
||||
build(): Promise<{
|
||||
incrementalAdminRouter: Router;
|
||||
}>;
|
||||
static create(
|
||||
env: PluginEnvironment,
|
||||
builder: CatalogBuilder,
|
||||
): Promise<IncrementalCatalogBuilder>;
|
||||
}
|
||||
|
||||
// @public
|
||||
export interface IncrementalEntityProvider<TCursor, TContext> {
|
||||
around(burst: (context: TContext) => Promise<void>): Promise<void>;
|
||||
getProviderName(): string;
|
||||
next(
|
||||
context: TContext,
|
||||
cursor?: TCursor,
|
||||
): Promise<EntityIteratorResult<TCursor>>;
|
||||
}
|
||||
|
||||
// @public (undocumented)
|
||||
export interface IncrementalEntityProviderOptions {
|
||||
backoff?: DurationObjectUnits[];
|
||||
burstInterval: DurationObjectUnits;
|
||||
burstLength: DurationObjectUnits;
|
||||
restLength: DurationObjectUnits;
|
||||
}
|
||||
|
||||
// @alpha
|
||||
export const incrementalIngestionEntityProviderCatalogModule: (options: {
|
||||
providers: {
|
||||
provider: IncrementalEntityProvider<unknown, unknown>;
|
||||
options: IncrementalEntityProviderOptions;
|
||||
}[];
|
||||
}) => BackendFeature;
|
||||
|
||||
// @public (undocumented)
|
||||
export type PluginEnvironment = {
|
||||
logger: Logger;
|
||||
database: PluginDatabaseManager;
|
||||
scheduler: PluginTaskScheduler;
|
||||
config: Config;
|
||||
reader: UrlReader;
|
||||
permissions: PermissionEvaluator;
|
||||
};
|
||||
```
|
||||
@@ -0,0 +1,26 @@
|
||||
/*
|
||||
* Copyright 2021 The Backstage Authors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
// This file makes it possible to run "yarn knex migrate:make some_file_name"
|
||||
// to assist in making new migrations
|
||||
module.exports = {
|
||||
client: 'better-sqlite3',
|
||||
connection: ':memory:',
|
||||
useNullAsDefault: true,
|
||||
migrations: {
|
||||
directory: './migrations',
|
||||
},
|
||||
};
|
||||
+177
@@ -0,0 +1,177 @@
|
||||
/*
|
||||
* Copyright 2022 The Backstage Authors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
/**
|
||||
* @param { import("knex").Knex } knex
|
||||
*/
|
||||
exports.up = async function up(knex) {
|
||||
/**
|
||||
* Sets up the ingestions table
|
||||
*/
|
||||
await knex.schema.createTable('ingestions', table => {
|
||||
table.comment('Tracks ingestion streams for very large data sets');
|
||||
|
||||
table
|
||||
.uuid('id', { primary: true })
|
||||
.notNullable()
|
||||
.comment('Auto-generated ID of the ingestion');
|
||||
|
||||
table
|
||||
.string('provider_name')
|
||||
.notNullable()
|
||||
.comment('each provider gets its own identifiable name');
|
||||
|
||||
table
|
||||
.string('status')
|
||||
.notNullable()
|
||||
.comment(
|
||||
'One of "interstitial" | "bursting" | "backing off" | "resting" | "complete"',
|
||||
);
|
||||
|
||||
table
|
||||
.string('next_action')
|
||||
.notNullable()
|
||||
.comment("what will this, 'ingest', 'rest', 'backoff', 'nothing (done)'");
|
||||
|
||||
table
|
||||
.timestamp('next_action_at')
|
||||
.notNullable()
|
||||
.defaultTo(knex.fn.now())
|
||||
.comment('the moment in time at which point ingestion can begin again');
|
||||
|
||||
table
|
||||
.string('last_error')
|
||||
.comment('records any error that occurred in the previous burst attempt');
|
||||
|
||||
table
|
||||
.integer('attempts')
|
||||
.notNullable()
|
||||
.defaultTo(0)
|
||||
.comment('how many attempts have been made to burst without success');
|
||||
|
||||
table
|
||||
.timestamp('created_at')
|
||||
.notNullable()
|
||||
.defaultTo(knex.fn.now())
|
||||
.comment('when did this ingestion actually begin');
|
||||
|
||||
table
|
||||
.timestamp('ingestion_completed_at')
|
||||
.comment('when did the ingestion actually end');
|
||||
|
||||
table
|
||||
.timestamp('rest_completed_at')
|
||||
.comment('when did the rest period actually end');
|
||||
|
||||
table
|
||||
.string('completion_ticket')
|
||||
.notNullable()
|
||||
.comment(
|
||||
'indicates whether the ticket is still open or stamped complete',
|
||||
);
|
||||
});
|
||||
|
||||
await knex.schema.alterTable('ingestions', t => {
|
||||
t.primary('id');
|
||||
t.index('provider_name', 'ingestion_provider_name_idx');
|
||||
t.unique(['provider_name', 'completion_ticket'], {
|
||||
indexName: 'ingestion_composite_index',
|
||||
deferrable: 'deferred',
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* Sets up the ingestion_marks table
|
||||
*/
|
||||
await knex.schema.createTable('ingestion_marks', table => {
|
||||
table.comment('tracks each step of an iterative ingestion');
|
||||
|
||||
table
|
||||
.uuid('id', { primary: true })
|
||||
.notNullable()
|
||||
.comment('Auto-generated ID of the ingestion mark');
|
||||
|
||||
table
|
||||
.uuid('ingestion_id')
|
||||
.notNullable()
|
||||
.references('id')
|
||||
.inTable('ingestions')
|
||||
.onDelete('CASCADE')
|
||||
.comment('The id of the ingestion in which this mark took place');
|
||||
|
||||
table
|
||||
.json('cursor')
|
||||
.comment(
|
||||
'the current data associated with this iteration wherever it is in this moment in time',
|
||||
);
|
||||
|
||||
table
|
||||
.integer('sequence')
|
||||
.notNullable()
|
||||
.defaultTo(0)
|
||||
.comment('what is the order of this mark');
|
||||
|
||||
table.timestamp('created_at').notNullable().defaultTo(knex.fn.now());
|
||||
});
|
||||
|
||||
await knex.schema.alterTable('ingestion_marks', t => {
|
||||
t.primary('id');
|
||||
t.index('ingestion_id', 'ingestion_mark_ingestion_id_idx');
|
||||
});
|
||||
|
||||
/**
|
||||
* Set up the ingestion_mark_entities table
|
||||
*/
|
||||
await knex.schema.createTable('ingestion_mark_entities', table => {
|
||||
table.comment(
|
||||
'tracks the entities recorded in each step of an iterative ingestion',
|
||||
);
|
||||
|
||||
table
|
||||
.uuid('id', { primary: true })
|
||||
.notNullable()
|
||||
.comment('Auto-generated ID of the marked entity');
|
||||
|
||||
table
|
||||
.uuid('ingestion_mark_id')
|
||||
.notNullable()
|
||||
.references('id')
|
||||
.inTable('ingestion_marks')
|
||||
.onDelete('CASCADE')
|
||||
.comment(
|
||||
'Every time a mark happens during an ingestion, there are a list of entities marked.',
|
||||
);
|
||||
|
||||
table
|
||||
.string('ref')
|
||||
.notNullable()
|
||||
.comment('the entity reference of the marked entity');
|
||||
});
|
||||
|
||||
await knex.schema.alterTable('ingestion_mark_entities', t => {
|
||||
t.primary('id');
|
||||
t.index('ingestion_mark_id', 'ingestion_mark_entity_ingestion_mark_id_idx');
|
||||
});
|
||||
};
|
||||
|
||||
/**
|
||||
* @param { import("knex").Knex } knex
|
||||
*/
|
||||
exports.down = async function down(knex) {
|
||||
await knex.schema.dropTable('ingestion_mark_entities');
|
||||
await knex.schema.dropTable('ingestion_marks');
|
||||
await knex.schema.dropTable('ingestions');
|
||||
};
|
||||
@@ -0,0 +1,67 @@
|
||||
{
|
||||
"name": "@backstage/plugin-catalog-backend-module-incremental-ingestion",
|
||||
"description": "An entity provider for streaming large asset sources into the catalog",
|
||||
"version": "0.0.0",
|
||||
"main": "src/index.ts",
|
||||
"types": "src/index.ts",
|
||||
"license": "Apache-2.0",
|
||||
"publishConfig": {
|
||||
"access": "public",
|
||||
"alphaTypes": "dist/index.alpha.d.ts",
|
||||
"main": "dist/index.cjs.js",
|
||||
"types": "dist/index.d.ts"
|
||||
},
|
||||
"backstage": {
|
||||
"role": "backend-plugin-module"
|
||||
},
|
||||
"homepage": "https://backstage.io",
|
||||
"repository": {
|
||||
"type": "git",
|
||||
"url": "https://github.com/backstage/backstage",
|
||||
"directory": "plugins/catalog-backend-module-incremental-ingestion"
|
||||
},
|
||||
"keywords": [
|
||||
"backstage"
|
||||
],
|
||||
"scripts": {
|
||||
"start": "backstage-cli package start",
|
||||
"build": "backstage-cli package build --experimental-type-build",
|
||||
"lint": "backstage-cli package lint",
|
||||
"test": "backstage-cli package test",
|
||||
"clean": "backstage-cli package clean",
|
||||
"prepack": "backstage-cli package prepack",
|
||||
"postpack": "backstage-cli package postpack"
|
||||
},
|
||||
"dependencies": {
|
||||
"@backstage/backend-common": "workspace:^",
|
||||
"@backstage/backend-plugin-api": "workspace:^",
|
||||
"@backstage/backend-tasks": "workspace:^",
|
||||
"@backstage/backend-test-utils": "workspace:^",
|
||||
"@backstage/catalog-model": "workspace:^",
|
||||
"@backstage/config": "workspace:^",
|
||||
"@backstage/errors": "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",
|
||||
"express": "^4.17.1",
|
||||
"express-promise-router": "^4.1.0",
|
||||
"knex": "^2.0.0",
|
||||
"lodash": "^4.17.21",
|
||||
"luxon": "^3.0.0",
|
||||
"uuid": "^8.3.2",
|
||||
"winston": "^3.2.1"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@backstage/backend-app-api": "workspace:^",
|
||||
"@backstage/cli": "workspace:^",
|
||||
"@backstage/plugin-catalog-backend": "workspace:^",
|
||||
"get-port": "^6.1.2"
|
||||
},
|
||||
"files": [
|
||||
"alpha",
|
||||
"dist",
|
||||
"migrations/**/*.{js,d.ts}"
|
||||
]
|
||||
}
|
||||
+587
@@ -0,0 +1,587 @@
|
||||
/*
|
||||
* Copyright 2022 The Backstage Authors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
import { Knex } from 'knex';
|
||||
import type { DeferredEntity } from '@backstage/plugin-catalog-backend';
|
||||
import { stringifyEntityRef } from '@backstage/catalog-model';
|
||||
import { Duration } from 'luxon';
|
||||
import { v4 } from 'uuid';
|
||||
import { INCREMENTAL_ENTITY_PROVIDER_ANNOTATION } from '../types';
|
||||
import {
|
||||
IngestionRecord,
|
||||
IngestionRecordUpdate,
|
||||
IngestionUpsert,
|
||||
MarkRecord,
|
||||
MarkRecordInsert,
|
||||
} from './tables';
|
||||
|
||||
export class IncrementalIngestionDatabaseManager {
|
||||
private client: Knex;
|
||||
|
||||
constructor(options: { client: Knex }) {
|
||||
this.client = options.client;
|
||||
}
|
||||
|
||||
/**
|
||||
* Performs an update to the ingestion record with matching `id`.
|
||||
* @param options - IngestionRecordUpdate
|
||||
*/
|
||||
async updateIngestionRecordById(options: IngestionRecordUpdate) {
|
||||
await this.client.transaction(async tx => {
|
||||
const { ingestionId, update } = options;
|
||||
await tx('ingestions').where('id', ingestionId).update(update);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Performs an update to the ingestion record with matching provider name. Will only update active records.
|
||||
* @param provider - string
|
||||
* @param update - Partial<IngestionUpsertIFace>
|
||||
*/
|
||||
async updateIngestionRecordByProvider(
|
||||
provider: string,
|
||||
update: Partial<IngestionUpsert>,
|
||||
) {
|
||||
await this.client.transaction(async tx => {
|
||||
await tx('ingestions')
|
||||
.where('provider_name', provider)
|
||||
.andWhere('completion_ticket', 'open')
|
||||
.update(update);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Performs an insert into the `ingestions` table with the supplied values.
|
||||
* @param record - IngestionUpsertIFace
|
||||
*/
|
||||
async insertIngestionRecord(record: IngestionUpsert) {
|
||||
await this.client.transaction(async tx => {
|
||||
await tx('ingestions').insert(record);
|
||||
});
|
||||
}
|
||||
|
||||
private async deleteMarkEntities(
|
||||
tx: Knex.Transaction,
|
||||
ids: { id: string }[],
|
||||
) {
|
||||
const chunks: { id: string }[][] = [];
|
||||
for (let i = 0; i < ids.length; i += 100) {
|
||||
const chunk = ids.slice(i, i + 100);
|
||||
chunks.push(chunk);
|
||||
}
|
||||
|
||||
let deleted = 0;
|
||||
|
||||
for (const chunk of chunks) {
|
||||
const chunkDeleted = await tx('ingestion_mark_entities')
|
||||
.delete()
|
||||
.whereIn(
|
||||
'id',
|
||||
chunk.map(entry => entry.id),
|
||||
);
|
||||
deleted += chunkDeleted;
|
||||
}
|
||||
|
||||
return deleted;
|
||||
}
|
||||
|
||||
/**
|
||||
* Finds the current ingestion record for the named provider.
|
||||
* @param provider - string
|
||||
* @returns IngestionRecord | undefined
|
||||
*/
|
||||
async getCurrentIngestionRecord(provider: string) {
|
||||
return await this.client.transaction(async tx => {
|
||||
const record = await tx<IngestionRecord>('ingestions')
|
||||
.where('provider_name', provider)
|
||||
.andWhere('completion_ticket', 'open')
|
||||
.first();
|
||||
return record;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Removes all entries from `ingestion_marks_entities`, `ingestion_marks`, and `ingestions`
|
||||
* for prior ingestions that completed (i.e., have a `completion_ticket` value other than 'open').
|
||||
* @param provider - string
|
||||
* @returns A count of deletions for each record type.
|
||||
*/
|
||||
async clearFinishedIngestions(provider: string) {
|
||||
return await this.client.transaction(async tx => {
|
||||
const markEntitiesDeleted = await tx('ingestion_mark_entities')
|
||||
.delete()
|
||||
.whereIn(
|
||||
'ingestion_mark_id',
|
||||
tx('ingestion_marks')
|
||||
.select('id')
|
||||
.whereIn(
|
||||
'ingestion_id',
|
||||
tx('ingestions')
|
||||
.select('id')
|
||||
.where('provider_name', provider)
|
||||
.andWhereNot('completion_ticket', 'open'),
|
||||
),
|
||||
);
|
||||
|
||||
const marksDeleted = await tx('ingestion_marks')
|
||||
.delete()
|
||||
.whereIn(
|
||||
'ingestion_id',
|
||||
tx('ingestions')
|
||||
.select('id')
|
||||
.where('provider_name', provider)
|
||||
.andWhereNot('completion_ticket', 'open'),
|
||||
);
|
||||
|
||||
const ingestionsDeleted = await tx('ingestions')
|
||||
.delete()
|
||||
.where('provider_name', provider)
|
||||
.andWhereNot('completion_ticket', 'open');
|
||||
|
||||
return {
|
||||
deletions: {
|
||||
markEntitiesDeleted,
|
||||
marksDeleted,
|
||||
ingestionsDeleted,
|
||||
},
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Automatically cleans up duplicate ingestion records if they were accidentally created.
|
||||
* Any ingestion record where the `rest_completed_at` is null (meaning it is active) AND
|
||||
* the ingestionId is incorrect is a duplicate ingestion record.
|
||||
* @param ingestionId - string
|
||||
* @param provider - string
|
||||
*/
|
||||
async clearDuplicateIngestions(ingestionId: string, provider: string) {
|
||||
await this.client.transaction(async tx => {
|
||||
const invalid = await tx<IngestionRecord>('ingestions')
|
||||
.where('provider_name', provider)
|
||||
.andWhere('rest_completed_at', null)
|
||||
.andWhereNot('id', ingestionId);
|
||||
|
||||
if (invalid.length > 0) {
|
||||
await tx('ingestions').delete().whereIn('id', invalid);
|
||||
await tx('ingestion_mark_entities')
|
||||
.delete()
|
||||
.whereIn(
|
||||
'ingestion_mark_id',
|
||||
tx('ingestion_marks').select('id').whereIn('ingestion_id', invalid),
|
||||
);
|
||||
await tx('ingestion_marks').delete().whereIn('ingestion_id', invalid);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* This method fully purges and resets all ingestion records for the named provider, and
|
||||
* leaves it in a paused state.
|
||||
* @param provider - string
|
||||
* @returns Counts of all deleted ingestion records
|
||||
*/
|
||||
async purgeAndResetProvider(provider: string) {
|
||||
return await this.client.transaction(async tx => {
|
||||
const ingestionIDs: { id: string }[] = await tx('ingestions')
|
||||
.select('id')
|
||||
.where('provider_name', provider);
|
||||
|
||||
const markIDs: { id: string }[] =
|
||||
ingestionIDs.length > 0
|
||||
? await tx('ingestion_marks')
|
||||
.select('id')
|
||||
.whereIn(
|
||||
'ingestion_id',
|
||||
ingestionIDs.map(entry => entry.id),
|
||||
)
|
||||
: [];
|
||||
|
||||
const markEntityIDs: { id: string }[] =
|
||||
markIDs.length > 0
|
||||
? await tx('ingestion_mark_entities')
|
||||
.select('id')
|
||||
.whereIn(
|
||||
'ingestion_mark_id',
|
||||
markIDs.map(entry => entry.id),
|
||||
)
|
||||
: [];
|
||||
|
||||
const markEntitiesDeleted = await this.deleteMarkEntities(
|
||||
tx,
|
||||
markEntityIDs,
|
||||
);
|
||||
|
||||
const marksDeleted =
|
||||
markIDs.length > 0
|
||||
? await tx('ingestion_marks')
|
||||
.delete()
|
||||
.whereIn(
|
||||
'ingestion_id',
|
||||
ingestionIDs.map(entry => entry.id),
|
||||
)
|
||||
: 0;
|
||||
|
||||
const ingestionsDeleted = await tx('ingestions')
|
||||
.delete()
|
||||
.where('provider_name', provider);
|
||||
|
||||
const next_action_at = new Date();
|
||||
next_action_at.setTime(next_action_at.getTime() + 24 * 60 * 60 * 1000);
|
||||
|
||||
await this.insertIngestionRecord({
|
||||
id: v4(),
|
||||
next_action: 'rest',
|
||||
provider_name: provider,
|
||||
next_action_at,
|
||||
ingestion_completed_at: new Date(),
|
||||
status: 'resting',
|
||||
completion_ticket: 'open',
|
||||
});
|
||||
|
||||
return { provider, ingestionsDeleted, marksDeleted, markEntitiesDeleted };
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new ingestion record.
|
||||
* @param provider - string
|
||||
* @returns A new ingestion record
|
||||
*/
|
||||
async createProviderIngestionRecord(provider: string) {
|
||||
const ingestionId = v4();
|
||||
const nextAction = 'ingest';
|
||||
try {
|
||||
await this.insertIngestionRecord({
|
||||
id: ingestionId,
|
||||
next_action: nextAction,
|
||||
provider_name: provider,
|
||||
status: 'bursting',
|
||||
completion_ticket: 'open',
|
||||
});
|
||||
return { ingestionId, nextAction, attempts: 0, nextActionAt: Date.now() };
|
||||
} catch (_e) {
|
||||
// Creating the ingestion record failed. Return undefined.
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Computes which entities to remove, if any, at the end of a burst.
|
||||
* @param provider - string
|
||||
* @param ingestionId - string
|
||||
* @returns All entities to remove for this burst.
|
||||
*/
|
||||
async computeRemoved(provider: string, ingestionId: string) {
|
||||
return await this.client.transaction(async tx => {
|
||||
const removed: { entity: string; ref: string }[] = await tx(
|
||||
'final_entities',
|
||||
)
|
||||
.select(
|
||||
tx.ref('final_entity').as('entity'),
|
||||
tx.ref('refresh_state.entity_ref').as('ref'),
|
||||
)
|
||||
.join(
|
||||
'refresh_state',
|
||||
'refresh_state.entity_id',
|
||||
'final_entities.entity_id',
|
||||
)
|
||||
.join('search', 'search.entity_id', 'final_entities.entity_id')
|
||||
.whereNotIn(
|
||||
'entity_ref',
|
||||
tx('ingestion_marks')
|
||||
.join(
|
||||
'ingestion_mark_entities',
|
||||
'ingestion_marks.id',
|
||||
'ingestion_mark_entities.ingestion_mark_id',
|
||||
)
|
||||
.select('ingestion_mark_entities.ref')
|
||||
.where('ingestion_marks.ingestion_id', ingestionId),
|
||||
)
|
||||
.andWhere(
|
||||
'search.key',
|
||||
`metadata.annotations.${INCREMENTAL_ENTITY_PROVIDER_ANNOTATION}`,
|
||||
)
|
||||
.andWhere('search.value', provider);
|
||||
return removed.map(entity => {
|
||||
return { entity: JSON.parse(entity.entity) };
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Performs a lookup of all providers that have duplicate active ingestion records.
|
||||
* @returns An array of all duplicate active ingestions
|
||||
*/
|
||||
async healthcheck() {
|
||||
return await this.client.transaction(async tx => {
|
||||
const records = await tx<{ id: string; provider_name: string }>(
|
||||
'ingestions',
|
||||
)
|
||||
.distinct('id', 'provider_name')
|
||||
.where('rest_completed_at', null);
|
||||
return records;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Skips any wait time for the next action to run.
|
||||
* @param provider - string
|
||||
*/
|
||||
async triggerNextProviderAction(provider: string) {
|
||||
await this.updateIngestionRecordByProvider(provider, {
|
||||
next_action_at: new Date(),
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Purges the following tables:
|
||||
* * `ingestions`
|
||||
* * `ingestion_marks`
|
||||
* * `ingestion_mark_entities`
|
||||
*
|
||||
* This function leaves the ingestions table with all providers in a paused state.
|
||||
* @returns Results from cleaning up all ingestion tables.
|
||||
*/
|
||||
async cleanupProviders() {
|
||||
const providers = await this.listProviders();
|
||||
|
||||
const ingestionsDeleted = await this.purgeTable('ingestions');
|
||||
|
||||
const next_action_at = new Date();
|
||||
next_action_at.setTime(next_action_at.getTime() + 24 * 60 * 60 * 1000);
|
||||
|
||||
for (const provider of providers) {
|
||||
await this.insertIngestionRecord({
|
||||
id: v4(),
|
||||
next_action: 'rest',
|
||||
provider_name: provider,
|
||||
next_action_at,
|
||||
ingestion_completed_at: new Date(),
|
||||
status: 'resting',
|
||||
completion_ticket: 'open',
|
||||
});
|
||||
}
|
||||
|
||||
const ingestionMarksDeleted = await this.purgeTable('ingestion_marks');
|
||||
const markEntitiesDeleted = await this.purgeTable(
|
||||
'ingestion_mark_entities',
|
||||
);
|
||||
|
||||
return { ingestionsDeleted, ingestionMarksDeleted, markEntitiesDeleted };
|
||||
}
|
||||
|
||||
/**
|
||||
* Configures the current ingestion record to ingest a burst.
|
||||
* @param ingestionId - string
|
||||
*/
|
||||
async setProviderIngesting(ingestionId: string) {
|
||||
await this.updateIngestionRecordById({
|
||||
ingestionId,
|
||||
update: { next_action: 'ingest' },
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Indicates the provider is currently ingesting a burst.
|
||||
* @param ingestionId - string
|
||||
*/
|
||||
async setProviderBursting(ingestionId: string) {
|
||||
await this.updateIngestionRecordById({
|
||||
ingestionId,
|
||||
update: { status: 'bursting' },
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Finalizes the current ingestion record to indicate that the post-ingestion rest period is complete.
|
||||
* @param ingestionId - string
|
||||
*/
|
||||
async setProviderComplete(ingestionId: string) {
|
||||
await this.updateIngestionRecordById({
|
||||
ingestionId,
|
||||
update: {
|
||||
next_action: 'nothing (done)',
|
||||
rest_completed_at: new Date(),
|
||||
status: 'complete',
|
||||
completion_ticket: v4(),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Marks ingestion as complete and starts the post-ingestion rest cycle.
|
||||
* @param ingestionId - string
|
||||
* @param restLength - Duration
|
||||
*/
|
||||
async setProviderResting(ingestionId: string, restLength: Duration) {
|
||||
await this.updateIngestionRecordById({
|
||||
ingestionId,
|
||||
update: {
|
||||
next_action: 'rest',
|
||||
next_action_at: new Date(Date.now() + restLength.as('milliseconds')),
|
||||
ingestion_completed_at: new Date(),
|
||||
status: 'resting',
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Marks ingestion as paused after a burst completes.
|
||||
* @param ingestionId - string
|
||||
*/
|
||||
async setProviderInterstitial(ingestionId: string) {
|
||||
await this.updateIngestionRecordById({
|
||||
ingestionId,
|
||||
update: { attempts: 0, status: 'interstitial' },
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Starts the cancel process for the current ingestion.
|
||||
* @param ingestionId - string
|
||||
* @param message - string (optional)
|
||||
*/
|
||||
async setProviderCanceling(ingestionId: string, message?: string) {
|
||||
const update: Partial<IngestionUpsert> = {
|
||||
next_action: 'cancel',
|
||||
last_error: message ? message : undefined,
|
||||
next_action_at: new Date(),
|
||||
status: 'canceling',
|
||||
};
|
||||
await this.updateIngestionRecordById({ ingestionId, update });
|
||||
}
|
||||
|
||||
/**
|
||||
* Completes the cancel process and triggers a new ingestion.
|
||||
* @param ingestionId - string
|
||||
*/
|
||||
async setProviderCanceled(ingestionId: string) {
|
||||
await this.updateIngestionRecordById({
|
||||
ingestionId,
|
||||
update: {
|
||||
next_action: 'nothing (canceled)',
|
||||
rest_completed_at: new Date(),
|
||||
status: 'complete',
|
||||
completion_ticket: v4(),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Configures the current ingestion to wait and retry, due to a data source error.
|
||||
* @param ingestionId - string
|
||||
* @param attempts - number
|
||||
* @param error - Error
|
||||
* @param backoffLength - number
|
||||
*/
|
||||
async setProviderBackoff(
|
||||
ingestionId: string,
|
||||
attempts: number,
|
||||
error: Error,
|
||||
backoffLength: number,
|
||||
) {
|
||||
await this.updateIngestionRecordById({
|
||||
ingestionId,
|
||||
update: {
|
||||
next_action: 'backoff',
|
||||
attempts: attempts + 1,
|
||||
last_error: String(error),
|
||||
next_action_at: new Date(Date.now() + backoffLength),
|
||||
status: 'backing off',
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the last record from `ingestion_marks` for the supplied ingestionId.
|
||||
* @param ingestionId - string
|
||||
* @returns MarkRecord | undefined
|
||||
*/
|
||||
async getLastMark(ingestionId: string) {
|
||||
return await this.client.transaction(async tx => {
|
||||
const mark = await tx<MarkRecord>('ingestion_marks')
|
||||
.where('ingestion_id', ingestionId)
|
||||
.orderBy('sequence', 'desc')
|
||||
.first();
|
||||
return mark;
|
||||
});
|
||||
}
|
||||
|
||||
async getAllMarks(ingestionId: string) {
|
||||
return await this.client.transaction(async tx => {
|
||||
const marks = await tx<MarkRecord>('ingestion_marks')
|
||||
.where('ingestion_id', ingestionId)
|
||||
.orderBy('sequence', 'desc');
|
||||
return marks;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Performs an insert into the `ingestion_marks` table with the supplied values.
|
||||
* @param options - MarkRecordInsert
|
||||
*/
|
||||
async createMark(options: MarkRecordInsert) {
|
||||
const { record } = options;
|
||||
await this.client.transaction(async tx => {
|
||||
await tx('ingestion_marks').insert(record);
|
||||
});
|
||||
}
|
||||
/**
|
||||
* Performs an upsert to the `ingestion_mark_entities` table for all deferred entities.
|
||||
* @param markId - string
|
||||
* @param entities - DeferredEntity[]
|
||||
*/
|
||||
async createMarkEntities(markId: string, entities: DeferredEntity[]) {
|
||||
await this.client.transaction(async tx => {
|
||||
await tx('ingestion_mark_entities').insert(
|
||||
entities.map(entity => ({
|
||||
id: v4(),
|
||||
ingestion_mark_id: markId,
|
||||
ref: stringifyEntityRef(entity.entity),
|
||||
})),
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Deletes the entire content of a table, and returns the number of records deleted.
|
||||
* @param table - string
|
||||
* @returns number
|
||||
*/
|
||||
async purgeTable(table: string) {
|
||||
return await this.client.transaction(async tx => {
|
||||
return await tx(table).delete();
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns a list of all providers.
|
||||
* @returns string[]
|
||||
*/
|
||||
async listProviders() {
|
||||
return await this.client.transaction(async tx => {
|
||||
const providers = await tx<{ provider_name: string }>(
|
||||
'ingestions',
|
||||
).distinct('provider_name');
|
||||
return providers.map(entry => entry.provider_name);
|
||||
});
|
||||
}
|
||||
|
||||
async updateByName(provider: string, update: Partial<IngestionUpsert>) {
|
||||
await this.updateIngestionRecordByProvider(provider, update);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
/*
|
||||
* Copyright 2022 The Backstage Authors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
import { resolvePackagePath } from '@backstage/backend-common';
|
||||
import { Knex } from 'knex';
|
||||
import { DB_MIGRATIONS_TABLE } from './tables';
|
||||
|
||||
export async function applyDatabaseMigrations(knex: Knex): Promise<void> {
|
||||
const migrationsDir = resolvePackagePath(
|
||||
'@backstage/plugin-catalog-backend-module-incremental-ingestion',
|
||||
'migrations',
|
||||
);
|
||||
|
||||
await knex.migrate.latest({
|
||||
directory: migrationsDir,
|
||||
tableName: DB_MIGRATIONS_TABLE,
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,118 @@
|
||||
/*
|
||||
* Copyright 2021 The Backstage Authors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
export const DB_MIGRATIONS_TABLE = 'incremental_ingestion__knex_migrations';
|
||||
|
||||
/**
|
||||
* The shape of data inserted into or updated in the `ingestions` table.
|
||||
*/
|
||||
export interface IngestionUpsert {
|
||||
/**
|
||||
* The ingestion record id.
|
||||
*/
|
||||
id?: string;
|
||||
/**
|
||||
* The next action the incremental entity provider will take.
|
||||
*/
|
||||
next_action:
|
||||
| 'rest'
|
||||
| 'ingest'
|
||||
| 'backoff'
|
||||
| 'cancel'
|
||||
| 'nothing (done)'
|
||||
| 'nothing (canceled)';
|
||||
/**
|
||||
* Current status of the incremental entity provider.
|
||||
*/
|
||||
status:
|
||||
| 'complete'
|
||||
| 'bursting'
|
||||
| 'resting'
|
||||
| 'canceling'
|
||||
| 'interstitial'
|
||||
| 'backing off';
|
||||
/**
|
||||
* The name of the incremental entity provider being updated.
|
||||
*/
|
||||
provider_name: string;
|
||||
/**
|
||||
* Date/time stamp for when the next action will trigger.
|
||||
*/
|
||||
next_action_at?: Date;
|
||||
/**
|
||||
* A record of the last error generated by the incremental entity provider.
|
||||
*/
|
||||
last_error?: string | null;
|
||||
/**
|
||||
* The number of attempts the provider has attempted during the current cycle.
|
||||
*/
|
||||
attempts?: number;
|
||||
/**
|
||||
* Date/time stamp for the completion of ingestion.
|
||||
*/
|
||||
ingestion_completed_at?: Date | string | null;
|
||||
/**
|
||||
* Date/time stamp for the end of the rest cycle before the next ingestion.
|
||||
*/
|
||||
rest_completed_at?: Date | string | null;
|
||||
/**
|
||||
* A record of the finalized status of the ingestion record. Values are either 'open' or a uuid.
|
||||
*/
|
||||
completion_ticket: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* This interface is for updating an existing ingestion record.
|
||||
*/
|
||||
export interface IngestionRecordUpdate {
|
||||
ingestionId: string;
|
||||
update: Partial<IngestionUpsert>;
|
||||
}
|
||||
|
||||
/**
|
||||
* The expected response from the `ingestion_marks` table.
|
||||
*/
|
||||
export interface MarkRecord {
|
||||
id: string;
|
||||
sequence: number;
|
||||
ingestion_id: string;
|
||||
cursor: string;
|
||||
created_at: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* The expected response from the `ingestions` table.
|
||||
*/
|
||||
export interface IngestionRecord extends IngestionUpsert {
|
||||
id: string;
|
||||
next_action_at: Date;
|
||||
/**
|
||||
* The date/time the ingestion record was created.
|
||||
*/
|
||||
created_at: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* This interface supplies all the values for adding an ingestion mark.
|
||||
*/
|
||||
export interface MarkRecordInsert {
|
||||
record: {
|
||||
id: string;
|
||||
ingestion_id: string;
|
||||
cursor: unknown;
|
||||
sequence: number;
|
||||
};
|
||||
}
|
||||
+296
@@ -0,0 +1,296 @@
|
||||
/*
|
||||
* Copyright 2022 The Backstage Authors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
import type { DeferredEntity } from '@backstage/plugin-catalog-backend';
|
||||
import {
|
||||
INCREMENTAL_ENTITY_PROVIDER_ANNOTATION,
|
||||
IterationEngine,
|
||||
IterationEngineOptions,
|
||||
} from '../types';
|
||||
import { IncrementalIngestionDatabaseManager } from '../database/IncrementalIngestionDatabaseManager';
|
||||
import type { AbortSignal } from 'node-abort-controller';
|
||||
import { performance } from 'perf_hooks';
|
||||
import { Duration, DurationObjectUnits } from 'luxon';
|
||||
import { v4 } from 'uuid';
|
||||
import { stringifyError } from '@backstage/errors';
|
||||
|
||||
export class IncrementalIngestionEngine implements IterationEngine {
|
||||
private readonly restLength: Duration;
|
||||
private readonly backoff: DurationObjectUnits[];
|
||||
|
||||
private manager: IncrementalIngestionDatabaseManager;
|
||||
|
||||
constructor(private options: IterationEngineOptions) {
|
||||
this.manager = options.manager;
|
||||
this.restLength = Duration.fromObject(options.restLength);
|
||||
this.backoff = options.backoff ?? [
|
||||
{ minutes: 1 },
|
||||
{ minutes: 5 },
|
||||
{ minutes: 30 },
|
||||
{ hours: 3 },
|
||||
];
|
||||
}
|
||||
|
||||
async taskFn(signal: AbortSignal) {
|
||||
try {
|
||||
this.options.logger.debug('Begin tick');
|
||||
await this.handleNextAction(signal);
|
||||
} catch (error) {
|
||||
this.options.logger.error(`${error}`);
|
||||
throw error;
|
||||
} finally {
|
||||
this.options.logger.debug('End tick');
|
||||
}
|
||||
}
|
||||
|
||||
async handleNextAction(signal: AbortSignal) {
|
||||
await this.options.ready;
|
||||
|
||||
const result = await this.getCurrentAction();
|
||||
if (result) {
|
||||
const { ingestionId, nextActionAt, nextAction, attempts } = result;
|
||||
|
||||
switch (nextAction) {
|
||||
case 'rest':
|
||||
if (Date.now() > nextActionAt) {
|
||||
await this.manager.clearFinishedIngestions(
|
||||
this.options.provider.getProviderName(),
|
||||
);
|
||||
this.options.logger.info(
|
||||
`incremental-engine: Ingestion ${ingestionId} rest period complete. Ingestion will start again`,
|
||||
);
|
||||
|
||||
await this.manager.setProviderComplete(ingestionId);
|
||||
} else {
|
||||
this.options.logger.info(
|
||||
`incremental-engine: Ingestion '${ingestionId}' rest period continuing`,
|
||||
);
|
||||
}
|
||||
break;
|
||||
case 'ingest':
|
||||
try {
|
||||
await this.manager.setProviderBursting(ingestionId);
|
||||
const done = await this.ingestOneBurst(ingestionId, signal);
|
||||
if (done) {
|
||||
this.options.logger.info(
|
||||
`incremental-engine: Ingestion '${ingestionId}' complete, transitioning to rest period of ${this.restLength.toHuman()}`,
|
||||
);
|
||||
await this.manager.setProviderResting(
|
||||
ingestionId,
|
||||
this.restLength,
|
||||
);
|
||||
} else {
|
||||
await this.manager.setProviderInterstitial(ingestionId);
|
||||
this.options.logger.info(
|
||||
`incremental-engine: Ingestion '${ingestionId}' continuing`,
|
||||
);
|
||||
}
|
||||
} catch (error) {
|
||||
if (
|
||||
(error as Error).message &&
|
||||
(error as Error).message === 'CANCEL'
|
||||
) {
|
||||
this.options.logger.info(
|
||||
`incremental-engine: Ingestion '${ingestionId}' canceled`,
|
||||
);
|
||||
await this.manager.setProviderCanceling(
|
||||
ingestionId,
|
||||
(error as Error).message,
|
||||
);
|
||||
} else {
|
||||
const currentBackoff = Duration.fromObject(
|
||||
this.backoff[Math.min(this.backoff.length - 1, attempts)],
|
||||
);
|
||||
|
||||
const backoffLength = currentBackoff.as('milliseconds');
|
||||
this.options.logger.error(error);
|
||||
|
||||
const truncatedError = stringifyError(error).substring(0, 700);
|
||||
this.options.logger.error(
|
||||
`incremental-engine: Ingestion '${ingestionId}' threw an error during ingestion burst. Ingestion will backoff for ${currentBackoff.toHuman()} (${truncatedError})`,
|
||||
);
|
||||
|
||||
await this.manager.setProviderBackoff(
|
||||
ingestionId,
|
||||
attempts,
|
||||
error as Error,
|
||||
backoffLength,
|
||||
);
|
||||
}
|
||||
}
|
||||
break;
|
||||
case 'backoff':
|
||||
if (Date.now() > nextActionAt) {
|
||||
this.options.logger.info(
|
||||
`incremental-engine: Ingestion '${ingestionId}' backoff complete, will attempt to resume`,
|
||||
);
|
||||
await this.manager.setProviderIngesting(ingestionId);
|
||||
} else {
|
||||
this.options.logger.info(
|
||||
`incremental-engine: Ingestion '${ingestionId}' backoff continuing`,
|
||||
);
|
||||
}
|
||||
break;
|
||||
case 'cancel':
|
||||
this.options.logger.info(
|
||||
`incremental-engine: Ingestion '${ingestionId}' canceling, will restart`,
|
||||
);
|
||||
await this.manager.setProviderCanceled(ingestionId);
|
||||
break;
|
||||
default:
|
||||
this.options.logger.error(
|
||||
`incremental-engine: Ingestion '${ingestionId}' received unknown action '${nextAction}'`,
|
||||
);
|
||||
}
|
||||
} else {
|
||||
this.options.logger.error(
|
||||
`incremental-engine: Engine tried to create duplicate ingestion record for provider '${this.options.provider.getProviderName()}'.`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
async getCurrentAction() {
|
||||
const providerName = this.options.provider.getProviderName();
|
||||
const record = await this.manager.getCurrentIngestionRecord(providerName);
|
||||
if (record) {
|
||||
this.options.logger.info(
|
||||
`incremental-engine: Ingestion record found: '${record.id}'`,
|
||||
);
|
||||
return {
|
||||
ingestionId: record.id,
|
||||
nextAction: record.next_action as 'rest' | 'ingest' | 'backoff',
|
||||
attempts: record.attempts as number,
|
||||
nextActionAt: record.next_action_at.valueOf() as number,
|
||||
};
|
||||
}
|
||||
const result = await this.manager.createProviderIngestionRecord(
|
||||
providerName,
|
||||
);
|
||||
if (result) {
|
||||
this.options.logger.info(
|
||||
`incremental-engine: Ingestion record created: '${result.ingestionId}'`,
|
||||
);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
async ingestOneBurst(id: string, signal: AbortSignal) {
|
||||
const lastMark = await this.manager.getLastMark(id);
|
||||
|
||||
const cursor = lastMark ? lastMark.cursor : undefined;
|
||||
let sequence = lastMark ? lastMark.sequence + 1 : 0;
|
||||
|
||||
const start = performance.now();
|
||||
let count = 0;
|
||||
let done = false;
|
||||
this.options.logger.info(
|
||||
`incremental-engine: Ingestion '${id}' burst initiated`,
|
||||
);
|
||||
|
||||
await this.options.provider.around(async (context: unknown) => {
|
||||
let next = await this.options.provider.next(context, cursor);
|
||||
count++;
|
||||
for (;;) {
|
||||
done = next.done;
|
||||
await this.mark({
|
||||
id,
|
||||
sequence,
|
||||
entities: next?.entities,
|
||||
done: next.done,
|
||||
cursor: next?.cursor,
|
||||
});
|
||||
if (signal.aborted || next.done) {
|
||||
break;
|
||||
} else {
|
||||
next = await this.options.provider.next(context, next.cursor);
|
||||
count++;
|
||||
sequence++;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
this.options.logger.info(
|
||||
`incremental-engine: Ingestion '${id}' burst complete. (${count} batches in ${Math.round(
|
||||
performance.now() - start,
|
||||
)}ms).`,
|
||||
);
|
||||
return done;
|
||||
}
|
||||
|
||||
async mark(options: {
|
||||
id: string;
|
||||
sequence: number;
|
||||
entities?: DeferredEntity[];
|
||||
done: boolean;
|
||||
cursor?: unknown;
|
||||
}) {
|
||||
const { id, sequence, entities, done, cursor } = options;
|
||||
this.options.logger.debug(
|
||||
`incremental-engine: Ingestion '${id}': MARK ${
|
||||
entities ? entities.length : 0
|
||||
} entities, cursor: ${
|
||||
cursor ? JSON.stringify(cursor) : 'none'
|
||||
}, done: ${done}`,
|
||||
);
|
||||
const markId = v4();
|
||||
|
||||
await this.manager.createMark({
|
||||
record: {
|
||||
id: markId,
|
||||
ingestion_id: id,
|
||||
cursor,
|
||||
sequence,
|
||||
},
|
||||
});
|
||||
|
||||
if (entities && entities.length > 0) {
|
||||
await this.manager.createMarkEntities(markId, entities);
|
||||
}
|
||||
|
||||
const added =
|
||||
entities?.map(deferred => ({
|
||||
...deferred,
|
||||
entity: {
|
||||
...deferred.entity,
|
||||
metadata: {
|
||||
...deferred.entity.metadata,
|
||||
annotations: {
|
||||
...deferred.entity.metadata.annotations,
|
||||
[INCREMENTAL_ENTITY_PROVIDER_ANNOTATION]:
|
||||
this.options.provider.getProviderName(),
|
||||
},
|
||||
},
|
||||
},
|
||||
})) ?? [];
|
||||
|
||||
const removed: DeferredEntity[] = [];
|
||||
|
||||
if (done) {
|
||||
removed.push(
|
||||
...(await this.manager.computeRemoved(
|
||||
this.options.provider.getProviderName(),
|
||||
id,
|
||||
)),
|
||||
);
|
||||
}
|
||||
|
||||
await this.options.connection.applyMutation({
|
||||
type: 'delta',
|
||||
added,
|
||||
removed,
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
/*
|
||||
* Copyright 2022 The Backstage Authors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* 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 * from './module';
|
||||
export * from './service';
|
||||
export {
|
||||
INCREMENTAL_ENTITY_PROVIDER_ANNOTATION,
|
||||
type EntityIteratorResult,
|
||||
type IncrementalEntityProvider,
|
||||
type IncrementalEntityProviderOptions,
|
||||
type PluginEnvironment,
|
||||
} from './types';
|
||||
+113
@@ -0,0 +1,113 @@
|
||||
/*
|
||||
* Copyright 2022 The Backstage Authors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
import { getVoidLogger } from '@backstage/backend-common';
|
||||
import { PluginTaskScheduler } from '@backstage/backend-tasks';
|
||||
import { TestDatabases } from '@backstage/backend-test-utils';
|
||||
import { ConfigReader } from '@backstage/config';
|
||||
import { IncrementalEntityProvider } from '../types';
|
||||
import { WrapperProviders } from './WrapperProviders';
|
||||
|
||||
describe('WrapperProviders', () => {
|
||||
const applyDatabaseMigrations = jest.fn();
|
||||
const databases = TestDatabases.create({
|
||||
ids: ['POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'],
|
||||
});
|
||||
const config = new ConfigReader({});
|
||||
const logger = getVoidLogger();
|
||||
const scheduler = {
|
||||
scheduleTask: jest.fn(),
|
||||
};
|
||||
|
||||
beforeEach(() => {
|
||||
jest.resetAllMocks();
|
||||
});
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should initialize the providers in order, %p',
|
||||
async databaseId => {
|
||||
const client = await databases.init(databaseId);
|
||||
|
||||
const provider1: IncrementalEntityProvider<number, {}> = {
|
||||
getProviderName: () => 'provider1',
|
||||
around: burst => burst(0),
|
||||
next: async (_context, cursor) => {
|
||||
return !cursor
|
||||
? { done: false, entities: [], cursor: 1 }
|
||||
: { done: true };
|
||||
},
|
||||
};
|
||||
|
||||
const provider2: IncrementalEntityProvider<number, {}> = {
|
||||
getProviderName: () => 'provider2',
|
||||
around: burst => burst(0),
|
||||
next: async (_context, cursor) => {
|
||||
return !cursor
|
||||
? { done: false, entities: [], cursor: 1 }
|
||||
: { done: true };
|
||||
},
|
||||
};
|
||||
|
||||
const providers = new WrapperProviders({
|
||||
config,
|
||||
logger,
|
||||
client,
|
||||
scheduler:
|
||||
scheduler as Partial<PluginTaskScheduler> as PluginTaskScheduler,
|
||||
applyDatabaseMigrations,
|
||||
});
|
||||
const wrapped1 = providers.wrap(provider1, {
|
||||
burstInterval: { seconds: 1 },
|
||||
burstLength: { seconds: 1 },
|
||||
restLength: { seconds: 1 },
|
||||
});
|
||||
const wrapped2 = providers.wrap(provider2, {
|
||||
burstInterval: { seconds: 1 },
|
||||
burstLength: { seconds: 1 },
|
||||
restLength: { seconds: 1 },
|
||||
});
|
||||
|
||||
let resolved = false;
|
||||
(providers as any).readySignal.then(() => {
|
||||
resolved = true;
|
||||
});
|
||||
|
||||
expect(applyDatabaseMigrations).toHaveBeenCalledTimes(0);
|
||||
expect(resolved).toBe(false);
|
||||
expect(scheduler.scheduleTask).not.toHaveBeenCalled();
|
||||
|
||||
await wrapped1.connect({} as any); // simulates the catalog engine
|
||||
|
||||
expect(resolved).toBe(false);
|
||||
expect(applyDatabaseMigrations).toHaveBeenCalledTimes(1);
|
||||
expect(scheduler.scheduleTask).toHaveBeenLastCalledWith(
|
||||
expect.objectContaining({
|
||||
id: 'provider1',
|
||||
}),
|
||||
);
|
||||
|
||||
await wrapped2.connect({} as any);
|
||||
|
||||
expect(resolved).toBe(true);
|
||||
expect(applyDatabaseMigrations).toHaveBeenCalledTimes(1);
|
||||
expect(scheduler.scheduleTask).toHaveBeenLastCalledWith(
|
||||
expect.objectContaining({
|
||||
id: 'provider2',
|
||||
}),
|
||||
);
|
||||
},
|
||||
);
|
||||
});
|
||||
@@ -0,0 +1,142 @@
|
||||
/*
|
||||
* Copyright 2022 The Backstage Authors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
import { Logger, loggerToWinstonLogger } from '@backstage/backend-plugin-api';
|
||||
import { PluginTaskScheduler } from '@backstage/backend-tasks';
|
||||
import { Config } from '@backstage/config';
|
||||
import { stringifyError } from '@backstage/errors';
|
||||
import {
|
||||
EntityProvider,
|
||||
EntityProviderConnection,
|
||||
} from '@backstage/plugin-catalog-node';
|
||||
import express from 'express';
|
||||
import { Knex } from 'knex';
|
||||
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 {
|
||||
IncrementalEntityProvider,
|
||||
IncrementalEntityProviderOptions,
|
||||
} from '../types';
|
||||
import { Deferred } from '../util';
|
||||
|
||||
/**
|
||||
* Helps in the creation of the catalog entity providers that wrap the
|
||||
* incremental ones.
|
||||
*/
|
||||
export class WrapperProviders {
|
||||
private migrate: Promise<void> | undefined;
|
||||
private numberOfProvidersToConnect = 0;
|
||||
private readonly readySignal = new Deferred<void>();
|
||||
|
||||
constructor(
|
||||
private readonly options: {
|
||||
config: Config;
|
||||
logger: Logger;
|
||||
client: Knex;
|
||||
scheduler: PluginTaskScheduler;
|
||||
applyDatabaseMigrations?: typeof applyDatabaseMigrations;
|
||||
},
|
||||
) {}
|
||||
|
||||
wrap(
|
||||
provider: IncrementalEntityProvider<unknown, unknown>,
|
||||
options: IncrementalEntityProviderOptions,
|
||||
): EntityProvider {
|
||||
this.numberOfProvidersToConnect += 1;
|
||||
return {
|
||||
getProviderName: () => provider.getProviderName(),
|
||||
connect: async connection => {
|
||||
await this.startProvider(provider, options, connection);
|
||||
this.numberOfProvidersToConnect -= 1;
|
||||
if (this.numberOfProvidersToConnect === 0) {
|
||||
this.readySignal.resolve();
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async adminRouter(): Promise<express.Router> {
|
||||
return createIncrementalProviderRouter(
|
||||
new IncrementalIngestionDatabaseManager({ client: this.options.client }),
|
||||
loggerToWinstonLogger(this.options.logger),
|
||||
);
|
||||
}
|
||||
|
||||
private async startProvider(
|
||||
provider: IncrementalEntityProvider<unknown, unknown>,
|
||||
providerOptions: IncrementalEntityProviderOptions,
|
||||
connection: EntityProviderConnection,
|
||||
) {
|
||||
const logger = loggerToWinstonLogger(
|
||||
this.options.logger.child({
|
||||
entityProvider: provider.getProviderName(),
|
||||
}),
|
||||
);
|
||||
|
||||
try {
|
||||
if (!this.migrate) {
|
||||
this.migrate = Promise.resolve().then(async () => {
|
||||
const apply =
|
||||
this.options.applyDatabaseMigrations ?? applyDatabaseMigrations;
|
||||
await apply(this.options.client);
|
||||
});
|
||||
}
|
||||
|
||||
await this.migrate;
|
||||
|
||||
const { burstInterval, burstLength, restLength } = providerOptions;
|
||||
|
||||
logger.info(`Connecting`);
|
||||
|
||||
const manager = new IncrementalIngestionDatabaseManager({
|
||||
client: this.options.client,
|
||||
});
|
||||
const engine = new IncrementalIngestionEngine({
|
||||
...providerOptions,
|
||||
ready: this.readySignal,
|
||||
manager,
|
||||
logger,
|
||||
provider,
|
||||
restLength,
|
||||
connection,
|
||||
});
|
||||
|
||||
const frequency = Duration.isDuration(burstInterval)
|
||||
? burstInterval
|
||||
: Duration.fromObject(burstInterval);
|
||||
const length = Duration.isDuration(burstLength)
|
||||
? burstLength
|
||||
: Duration.fromObject(burstLength);
|
||||
|
||||
await this.options.scheduler.scheduleTask({
|
||||
id: provider.getProviderName(),
|
||||
fn: engine.taskFn.bind(engine),
|
||||
frequency,
|
||||
timeout: length,
|
||||
});
|
||||
} catch (error) {
|
||||
logger.warn(
|
||||
`Failed to initialize incremental ingestion provider ${provider.getProviderName()}, ${stringifyError(
|
||||
error,
|
||||
)}`,
|
||||
);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
+87
@@ -0,0 +1,87 @@
|
||||
/*
|
||||
* Copyright 2022 The Backstage Authors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
import { getVoidLogger } from '@backstage/backend-common';
|
||||
import {
|
||||
configServiceRef,
|
||||
databaseServiceRef,
|
||||
httpRouterServiceRef,
|
||||
loggerServiceRef,
|
||||
schedulerServiceRef,
|
||||
} from '@backstage/backend-plugin-api';
|
||||
import { startTestBackend } from '@backstage/backend-test-utils';
|
||||
import { ConfigReader } from '@backstage/config';
|
||||
import { catalogProcessingExtensionPoint } from '@backstage/plugin-catalog-node';
|
||||
import { IncrementalEntityProvider } from '../types';
|
||||
import { incrementalIngestionEntityProviderCatalogModule } from './incrementalIngestionEntityProviderCatalogModule';
|
||||
|
||||
describe('bitbucketServerEntityProviderCatalogModule', () => {
|
||||
it('should register provider at the catalog extension point', async () => {
|
||||
const provider1: IncrementalEntityProvider<number, {}> = {
|
||||
getProviderName: () => 'provider1',
|
||||
around: burst => burst(0),
|
||||
next: async (cursor, _context) => {
|
||||
return !cursor
|
||||
? { done: false, entities: [], cursor: 1 }
|
||||
: { done: true };
|
||||
},
|
||||
};
|
||||
|
||||
const addEntityProvider = jest.fn();
|
||||
const httpRouterUse = jest.fn();
|
||||
|
||||
const scheduler = {};
|
||||
const database = {
|
||||
getClient: jest.fn(),
|
||||
};
|
||||
const httpRouter = {
|
||||
use: httpRouterUse,
|
||||
};
|
||||
|
||||
await startTestBackend({
|
||||
extensionPoints: [
|
||||
[catalogProcessingExtensionPoint, { addEntityProvider }],
|
||||
],
|
||||
services: [
|
||||
[configServiceRef, new ConfigReader({})],
|
||||
[databaseServiceRef, database],
|
||||
[httpRouterServiceRef, httpRouter],
|
||||
[loggerServiceRef, getVoidLogger()],
|
||||
[schedulerServiceRef, scheduler],
|
||||
],
|
||||
features: [
|
||||
incrementalIngestionEntityProviderCatalogModule({
|
||||
providers: [
|
||||
{
|
||||
provider: provider1,
|
||||
options: {
|
||||
burstInterval: { seconds: 1 },
|
||||
burstLength: { seconds: 1 },
|
||||
restLength: { seconds: 1 },
|
||||
},
|
||||
},
|
||||
],
|
||||
}),
|
||||
],
|
||||
});
|
||||
|
||||
expect(addEntityProvider).toHaveBeenCalledTimes(1);
|
||||
expect(addEntityProvider.mock.calls[0][0].getProviderName()).toBe(
|
||||
'provider1',
|
||||
);
|
||||
expect(httpRouterUse).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
+85
@@ -0,0 +1,85 @@
|
||||
/*
|
||||
* Copyright 2022 The Backstage Authors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
import {
|
||||
configServiceRef,
|
||||
createBackendModule,
|
||||
databaseServiceRef,
|
||||
httpRouterServiceRef,
|
||||
loggerServiceRef,
|
||||
schedulerServiceRef,
|
||||
} from '@backstage/backend-plugin-api';
|
||||
import { catalogProcessingExtensionPoint } from '@backstage/plugin-catalog-node';
|
||||
import {
|
||||
IncrementalEntityProvider,
|
||||
IncrementalEntityProviderOptions,
|
||||
} from '../types';
|
||||
import { WrapperProviders } from './WrapperProviders';
|
||||
|
||||
/**
|
||||
* Registers the incremental entity provider with the catalog processing extension point.
|
||||
*
|
||||
* @alpha
|
||||
*/
|
||||
export const incrementalIngestionEntityProviderCatalogModule =
|
||||
createBackendModule({
|
||||
pluginId: 'catalog',
|
||||
moduleId: 'incrementalIngestionEntityProvider',
|
||||
register(
|
||||
env,
|
||||
options: {
|
||||
providers: Array<{
|
||||
provider: IncrementalEntityProvider<unknown, unknown>;
|
||||
options: IncrementalEntityProviderOptions;
|
||||
}>;
|
||||
},
|
||||
) {
|
||||
env.registerInit({
|
||||
deps: {
|
||||
catalog: catalogProcessingExtensionPoint,
|
||||
config: configServiceRef,
|
||||
database: databaseServiceRef,
|
||||
httpRouter: httpRouterServiceRef,
|
||||
logger: loggerServiceRef,
|
||||
scheduler: schedulerServiceRef,
|
||||
},
|
||||
async init({
|
||||
catalog,
|
||||
config,
|
||||
database,
|
||||
httpRouter,
|
||||
logger,
|
||||
scheduler,
|
||||
}) {
|
||||
const client = await database.getClient();
|
||||
|
||||
const providers = new WrapperProviders({
|
||||
config,
|
||||
logger,
|
||||
client,
|
||||
scheduler,
|
||||
});
|
||||
|
||||
for (const entry of options.providers) {
|
||||
const wrapped = providers.wrap(entry.provider, entry.options);
|
||||
catalog.addEntityProvider(wrapped);
|
||||
}
|
||||
|
||||
httpRouter.use(await providers.adminRouter());
|
||||
},
|
||||
});
|
||||
},
|
||||
});
|
||||
@@ -0,0 +1,17 @@
|
||||
/*
|
||||
* Copyright 2022 The Backstage Authors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
export { incrementalIngestionEntityProviderCatalogModule } from './incrementalIngestionEntityProviderCatalogModule';
|
||||
@@ -0,0 +1,19 @@
|
||||
/*
|
||||
* Copyright 2022 The Backstage Authors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
export const PROVIDER_CLEANUP = '/incremental/cleanup';
|
||||
export const PROVIDER_HEALTH = '/incremental/health';
|
||||
export const PROVIDER_BASE_PATH = '/incremental/providers/:provider';
|
||||
@@ -0,0 +1,232 @@
|
||||
/*
|
||||
* Copyright 2022 The Backstage Authors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* 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 '../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());
|
||||
|
||||
// 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)),
|
||||
];
|
||||
|
||||
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 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 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)) {
|
||||
res.json({
|
||||
success: true,
|
||||
status: {
|
||||
current_action: 'rest complete, waiting to start',
|
||||
},
|
||||
});
|
||||
} else {
|
||||
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());
|
||||
|
||||
return router;
|
||||
};
|
||||
@@ -0,0 +1,99 @@
|
||||
/*
|
||||
* Copyright 2022 The Backstage Authors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
// eslint-disable-next-line import/no-extraneous-dependencies
|
||||
import {
|
||||
databaseFactory,
|
||||
discoveryFactory,
|
||||
httpRouterFactory,
|
||||
lifecycleFactory,
|
||||
loggerFactory,
|
||||
permissionsFactory,
|
||||
rootLoggerFactory,
|
||||
schedulerFactory,
|
||||
tokenManagerFactory,
|
||||
urlReaderFactory,
|
||||
} from '@backstage/backend-app-api';
|
||||
import { configServiceRef } from '@backstage/backend-plugin-api';
|
||||
import { startTestBackend } from '@backstage/backend-test-utils';
|
||||
import { ConfigReader } from '@backstage/config';
|
||||
import { catalogPlugin } from '@backstage/plugin-catalog-backend';
|
||||
import {
|
||||
IncrementalEntityProvider,
|
||||
incrementalIngestionEntityProviderCatalogModule,
|
||||
} from '.';
|
||||
|
||||
const provider: IncrementalEntityProvider<number, {}> = {
|
||||
getProviderName: () => 'test-provider',
|
||||
around: burst => burst(0),
|
||||
next: async (_context, cursor) => {
|
||||
await new Promise(resolve => setTimeout(resolve, 500));
|
||||
if (cursor === undefined || cursor < 3) {
|
||||
console.log(`### Returning batch #${cursor}`);
|
||||
return { done: false, entities: [], cursor: (cursor ?? 0) + 1 };
|
||||
}
|
||||
|
||||
console.log('### Last batch reached, stopping');
|
||||
return { done: true };
|
||||
},
|
||||
};
|
||||
|
||||
async function main() {
|
||||
const config = {
|
||||
backend: {
|
||||
baseUrl: 'http://localhost:7007',
|
||||
listen: ':7007',
|
||||
database: { client: 'better-sqlite3', connection: ':memory:' },
|
||||
},
|
||||
};
|
||||
|
||||
await startTestBackend({
|
||||
services: [
|
||||
[configServiceRef, new ConfigReader(config)],
|
||||
databaseFactory(),
|
||||
discoveryFactory(),
|
||||
httpRouterFactory(),
|
||||
lifecycleFactory(),
|
||||
loggerFactory(),
|
||||
permissionsFactory(),
|
||||
rootLoggerFactory(),
|
||||
schedulerFactory(),
|
||||
tokenManagerFactory(),
|
||||
urlReaderFactory(),
|
||||
],
|
||||
extensionPoints: [],
|
||||
features: [
|
||||
catalogPlugin(),
|
||||
incrementalIngestionEntityProviderCatalogModule({
|
||||
providers: [
|
||||
{
|
||||
provider: provider,
|
||||
options: {
|
||||
burstInterval: { seconds: 1 },
|
||||
burstLength: { seconds: 10 },
|
||||
restLength: { seconds: 10 },
|
||||
},
|
||||
},
|
||||
],
|
||||
}),
|
||||
],
|
||||
});
|
||||
}
|
||||
|
||||
main().catch(error => {
|
||||
console.error(error.stack);
|
||||
process.exit(1);
|
||||
});
|
||||
+116
@@ -0,0 +1,116 @@
|
||||
/*
|
||||
* Copyright 2022 The Backstage Authors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
import {
|
||||
IncrementalEntityProvider,
|
||||
IncrementalEntityProviderOptions,
|
||||
PluginEnvironment,
|
||||
} from '../types';
|
||||
import { CatalogBuilder as CoreCatalogBuilder } from '@backstage/plugin-catalog-backend';
|
||||
import { Duration } from 'luxon';
|
||||
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 { Deferred } from '../util';
|
||||
|
||||
/** @public */
|
||||
export class IncrementalCatalogBuilder {
|
||||
/**
|
||||
* Creates the incremental catalog builder, which extends the regular catalog builder.
|
||||
* @param env - PluginEnvironment
|
||||
* @param builder - CatalogBuilder
|
||||
* @returns IncrementalCatalogBuilder
|
||||
*/
|
||||
static async create(env: PluginEnvironment, builder: CoreCatalogBuilder) {
|
||||
const client = await env.database.getClient();
|
||||
const manager = new IncrementalIngestionDatabaseManager({ client });
|
||||
return new IncrementalCatalogBuilder(env, builder, client, manager);
|
||||
}
|
||||
|
||||
private ready: Deferred<void>;
|
||||
|
||||
private constructor(
|
||||
private env: PluginEnvironment,
|
||||
private builder: CoreCatalogBuilder,
|
||||
private client: Knex,
|
||||
private manager: IncrementalIngestionDatabaseManager,
|
||||
) {
|
||||
this.ready = new Deferred<void>();
|
||||
}
|
||||
|
||||
async build() {
|
||||
await applyDatabaseMigrations(this.client);
|
||||
this.ready.resolve();
|
||||
|
||||
const routerLogger = this.env.logger.child({
|
||||
router: 'IncrementalProviderAdmin',
|
||||
});
|
||||
|
||||
const incrementalAdminRouter = await createIncrementalProviderRouter(
|
||||
this.manager,
|
||||
routerLogger,
|
||||
);
|
||||
|
||||
return { incrementalAdminRouter };
|
||||
}
|
||||
|
||||
addIncrementalEntityProvider<TCursor, TContext>(
|
||||
provider: IncrementalEntityProvider<TCursor, TContext>,
|
||||
options: IncrementalEntityProviderOptions,
|
||||
) {
|
||||
const { burstInterval, burstLength, restLength } = options;
|
||||
const { logger: catalogLogger, scheduler } = this.env;
|
||||
const ready = this.ready;
|
||||
|
||||
const manager = this.manager;
|
||||
|
||||
this.builder.addEntityProvider({
|
||||
getProviderName: provider.getProviderName.bind(provider),
|
||||
async connect(connection) {
|
||||
const logger = catalogLogger.child({
|
||||
entityProvider: provider.getProviderName(),
|
||||
});
|
||||
|
||||
logger.info(`Connecting`);
|
||||
|
||||
const engine = new IncrementalIngestionEngine({
|
||||
...options,
|
||||
ready,
|
||||
manager,
|
||||
logger,
|
||||
provider,
|
||||
restLength,
|
||||
connection,
|
||||
});
|
||||
|
||||
const frequency = Duration.isDuration(burstInterval)
|
||||
? burstInterval
|
||||
: Duration.fromObject(burstInterval);
|
||||
const length = Duration.isDuration(burstLength)
|
||||
? burstLength
|
||||
: Duration.fromObject(burstLength);
|
||||
|
||||
await scheduler.scheduleTask({
|
||||
id: provider.getProviderName(),
|
||||
fn: engine.taskFn.bind(engine),
|
||||
frequency,
|
||||
timeout: length,
|
||||
});
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
/*
|
||||
* Copyright 2022 The Backstage Authors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
export { IncrementalCatalogBuilder } from './IncrementalCatalogBuilder';
|
||||
@@ -0,0 +1,157 @@
|
||||
/*
|
||||
* Copyright 2022 The Backstage Authors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
import type {
|
||||
PluginDatabaseManager,
|
||||
UrlReader,
|
||||
} from '@backstage/backend-common';
|
||||
import type {
|
||||
PluginTaskScheduler,
|
||||
TaskFunction,
|
||||
} from '@backstage/backend-tasks';
|
||||
import type { Config } from '@backstage/config';
|
||||
import type {
|
||||
DeferredEntity,
|
||||
EntityProviderConnection,
|
||||
} from '@backstage/plugin-catalog-backend';
|
||||
import type { PermissionEvaluator } from '@backstage/plugin-permission-common';
|
||||
import type { DurationObjectUnits } from 'luxon';
|
||||
import type { Logger } from 'winston';
|
||||
import { IncrementalIngestionDatabaseManager } from './database/IncrementalIngestionDatabaseManager';
|
||||
|
||||
/**
|
||||
* Entity annotation containing the incremental entity provider.
|
||||
*
|
||||
* @public
|
||||
*/
|
||||
export const INCREMENTAL_ENTITY_PROVIDER_ANNOTATION =
|
||||
'backstage.io/incremental-provider-name';
|
||||
|
||||
/**
|
||||
* Ingest entities into the catalog in bite-sized chunks.
|
||||
*
|
||||
* A Normal `EntityProvider` allows you to introduce entities into the
|
||||
* processing pipeline by calling an `applyMutation()` on the full set
|
||||
* of entities. However, this is not great when the number of entities
|
||||
* that you have to keep track of is extremely large because it
|
||||
* entails having all of them in memory at once. An
|
||||
* `IncrementalEntityProvider` by contrast allows you to provide
|
||||
* batches of entities in sequence so that you never need to have more
|
||||
* than a few hundred in memory at a time.
|
||||
*
|
||||
* @public
|
||||
*/
|
||||
export interface IncrementalEntityProvider<TCursor, TContext> {
|
||||
/**
|
||||
* This name must be unique between all of the entity providers
|
||||
* operating in the catalog.
|
||||
*/
|
||||
getProviderName(): string;
|
||||
|
||||
/**
|
||||
* Return a single page of entities from a specific point in the
|
||||
* ingestion.
|
||||
*
|
||||
* @param context - anything needed in order to fetch a single page.
|
||||
* @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.
|
||||
*/
|
||||
next(
|
||||
context: TContext,
|
||||
cursor?: TCursor,
|
||||
): Promise<EntityIteratorResult<TCursor>>;
|
||||
|
||||
/**
|
||||
* Do any setup and teardown necessary in order to provide the
|
||||
* context for fetching pages. This should always invoke `burst` in
|
||||
* order to fetch the individual pages.
|
||||
*
|
||||
* @param burst - a function which performs a series of iterations
|
||||
*/
|
||||
around(burst: (context: TContext) => Promise<void>): Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Value returned by an {@link IncrementalEntityProvider} to provide a
|
||||
* single page of entities to ingest.
|
||||
*
|
||||
* @public
|
||||
*/
|
||||
export type EntityIteratorResult<T> =
|
||||
| {
|
||||
done: false;
|
||||
entities: DeferredEntity[];
|
||||
cursor: T;
|
||||
}
|
||||
| {
|
||||
done: true;
|
||||
entities?: DeferredEntity[];
|
||||
cursor?: T;
|
||||
};
|
||||
|
||||
/** @public */
|
||||
export interface IncrementalEntityProviderOptions {
|
||||
/**
|
||||
* Entities are ingested in bursts. This interval determines how
|
||||
* much time to wait in between each burst.
|
||||
*/
|
||||
burstInterval: DurationObjectUnits;
|
||||
|
||||
/**
|
||||
* Entities are ingested in bursts. This value determines how long
|
||||
* to keep ingesting within each burst.
|
||||
*/
|
||||
burstLength: DurationObjectUnits;
|
||||
|
||||
/**
|
||||
* After a successful ingestion, the incremental entity provider
|
||||
* will rest for this period of time before starting to ingest
|
||||
* again.
|
||||
*/
|
||||
restLength: DurationObjectUnits;
|
||||
|
||||
/**
|
||||
* In the event of an error during an ingestion burst, the backoff
|
||||
* determines how soon it will be retried. E.g.
|
||||
* `[{ minutes: 1}, { minutes: 5}, {minutes: 30 }, { hours: 3 }]`
|
||||
*/
|
||||
backoff?: DurationObjectUnits[];
|
||||
}
|
||||
|
||||
/** @public */
|
||||
export type PluginEnvironment = {
|
||||
logger: Logger;
|
||||
database: PluginDatabaseManager;
|
||||
scheduler: PluginTaskScheduler;
|
||||
config: Config;
|
||||
reader: UrlReader;
|
||||
permissions: PermissionEvaluator;
|
||||
};
|
||||
|
||||
export interface IterationEngine {
|
||||
taskFn: TaskFunction;
|
||||
}
|
||||
|
||||
export interface IterationEngineOptions {
|
||||
logger: Logger;
|
||||
connection: EntityProviderConnection;
|
||||
manager: IncrementalIngestionDatabaseManager;
|
||||
provider: IncrementalEntityProvider<unknown, unknown>;
|
||||
restLength: DurationObjectUnits;
|
||||
ready: Promise<void>;
|
||||
backoff?: IncrementalEntityProviderOptions['backoff'];
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
/*
|
||||
* Copyright 2022 The Backstage Authors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
export class Deferred<T> implements Promise<T> {
|
||||
#resolve?: (value: T) => void;
|
||||
#reject?: (error: Error) => void;
|
||||
|
||||
public get resolve() {
|
||||
return this.#resolve!;
|
||||
}
|
||||
public get reject() {
|
||||
return this.#reject!;
|
||||
}
|
||||
|
||||
public then: Promise<T>['then'];
|
||||
public catch: Promise<T>['catch'];
|
||||
public finally: Promise<T>['finally'];
|
||||
|
||||
public constructor() {
|
||||
const promise = new Promise<T>((resolve, reject) => {
|
||||
this.#resolve = resolve;
|
||||
this.#reject = reject;
|
||||
});
|
||||
|
||||
this.then = promise.then.bind(promise);
|
||||
this.catch = promise.catch.bind(promise);
|
||||
this.finally = promise.finally.bind(promise);
|
||||
}
|
||||
|
||||
[Symbol.toStringTag]: 'Deferred' = 'Deferred';
|
||||
}
|
||||
Reference in New Issue
Block a user