Adding the incremental entity provider backend
Signed-off-by: Damon Kaswell <damon.kaswell1@hp.com>
This commit is contained in:
Damon Kaswell
committed by
Fredrik Adelöw
Fredrik Adelöw
parent
16cac2a684
commit
98c643a1a2
@@ -0,0 +1,5 @@
|
||||
---
|
||||
'@backstage/plugin-incremental-ingestion-backend': patch
|
||||
---
|
||||
|
||||
This change introduces incremental entity providers, which are used for streaming very large data sources into the catalog.
|
||||
@@ -44,6 +44,7 @@
|
||||
"@backstage/plugin-events-backend": "workspace:^",
|
||||
"@backstage/plugin-events-node": "workspace:^",
|
||||
"@backstage/plugin-graphql-backend": "workspace:^",
|
||||
"@backstage/plugin-incremental-ingestion-backend": "^0.1.0",
|
||||
"@backstage/plugin-jenkins-backend": "workspace:^",
|
||||
"@backstage/plugin-kafka-backend": "workspace:^",
|
||||
"@backstage/plugin-kubernetes-backend": "workspace:^",
|
||||
|
||||
@@ -0,0 +1,321 @@
|
||||
# backend-incremental-ingestion
|
||||
|
||||
The Incremental Ingestion Backend plugin provides an Incremental Entity Provider that can be used to ingest data from sources using delta mutations, while retaining the orphan prevention mechanism provided by full mutations.
|
||||
|
||||
## 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 `frontside/incremental-entity-provider: <entity-provider-id>` against all marked entities to determine which entities commited 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/backoff - Failed bursts are automatically retried with a built-in backoff 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 concequences:
|
||||
|
||||
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.
|
||||
|
||||
## Installation
|
||||
|
||||
1. Install `@backstage/plugin-incremental-ingestion-backend` with `yarn add @backstage/plugin-incremental-ingestion-backend`
|
||||
2. Import `IncrementalCatalogBuilder` from `@backstage/plugin-incremental-ingestion-backend` and instantiate it with `await IncrementalCatalogBuilder.create(env, builder)`. You have to pass `builder` into `IncrementalCatalogBuilder.create` function because `IncrementalCatalogBuilder` will convert an `IncrementalEntityProvider` into an `EntityProvider` and call `builder.addEntityProvider`.
|
||||
```ts
|
||||
const builder = CatalogBuilder.create(env);
|
||||
// incremental builder receives builder because it'll register
|
||||
// incremental entity providers with the builder
|
||||
const incrementalBuilder = await IncrementalCatalogBuilder.create(env, builder);
|
||||
```
|
||||
3. Last step, add `await incrementBuilder.build()` after `await builder.build()` to ensure that all `CatalogBuider` 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 '@frontside/backstage-plugin-incremental-ingestion-backend';
|
||||
import { GithubRepositoryEntityProvider } from '@frontside/backstage-plugin-incremental-ingestion-github';
|
||||
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
|
||||
await incrementalBuilder.build();
|
||||
// If you want to use a suite of administrative routes to control the incremental providers, use
|
||||
// `const { incrementalAdminRouter } = await incrementalBuilder.build();`
|
||||
|
||||
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 uniqiue 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-incremental-ingestion-backend';
|
||||
|
||||
// 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 teardown. 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 th client.
|
||||
|
||||
```ts
|
||||
export class MyIncrementalEntityProvider implements IncrementalEntityProvider<Cursor, Context> {
|
||||
|
||||
token: string;
|
||||
|
||||
construtor(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<Cursor, Context> {
|
||||
|
||||
token: string;
|
||||
|
||||
construtor(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<TCursor>> {
|
||||
const { apiClient } = context;
|
||||
const { page } = cursor;
|
||||
|
||||
// call your API with the current page
|
||||
const data = await apiClient.getServices(page);
|
||||
|
||||
// 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 = 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 })
|
||||
}
|
||||
)
|
||||
```
|
||||
|
||||
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 backoff period. It'll keep trying until it reaches the last backoff attempt. You don't need to do anything special to handle the retry logic.
|
||||
@@ -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',
|
||||
},
|
||||
};
|
||||
@@ -0,0 +1,90 @@
|
||||
exports.up = async function (knex) {
|
||||
await knex.raw(`
|
||||
CREATE VIEW ingestion.current_entities as SELECT
|
||||
FORMAT('%s:%s/%s',
|
||||
LOWER(final_entity::json #>> '{kind}'),
|
||||
LOWER(final_entity::json #>> '{metadata, namespace}'),
|
||||
LOWER(final_entity::json #>> '{metadata, name}')) as ref,
|
||||
final_entity::json #>> '{metadata, annotations, hp.com/provider-name}' as provider_name,
|
||||
final_entity FROM public.final_entities;
|
||||
`);
|
||||
|
||||
const schema = () => knex.schema.withSchema('ingestion');
|
||||
|
||||
await 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')
|
||||
.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 occured in the previous burst attempt');
|
||||
|
||||
table.integer('attempts').defaultTo(0).comment('how many attempts have been made to burst without success');
|
||||
|
||||
table.timestamp('created_at').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');
|
||||
});
|
||||
|
||||
await 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().comment('The id of the ingestion in which this mark took place');
|
||||
|
||||
// table
|
||||
// .foreign('ingestion_id').references('ingestions.id');
|
||||
|
||||
table
|
||||
.json('cursor')
|
||||
.comment('the current data associated with this iteration wherever it is in this moment in time');
|
||||
|
||||
table.integer('sequence').defaultTo(0).comment('what is the order of this mark');
|
||||
|
||||
table.timestamp('created_at').defaultTo(knex.fn.now());
|
||||
});
|
||||
|
||||
await 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()
|
||||
.comment('Every time a mark happens during an ingestion, there are a list of entities marked.');
|
||||
|
||||
// table
|
||||
// .foreign('ingestion_mark_id').references('ingestion_marks.id');
|
||||
|
||||
table.string('ref').notNullable().comment('the entity reference of the marked entity');
|
||||
});
|
||||
};
|
||||
|
||||
exports.down = async function (knex) {
|
||||
const schema = () => knex.schema.withSchema('ingestion');
|
||||
|
||||
await schema().dropView('current_entities');
|
||||
|
||||
await schema().dropTable('ingestion_mark_entities');
|
||||
|
||||
await schema().dropTable('ingestion_marks');
|
||||
|
||||
await schema().dropTable('ingestions');
|
||||
};
|
||||
+53
@@ -0,0 +1,53 @@
|
||||
/**
|
||||
* @param { import("knex").Knex } knex
|
||||
* @returns { Promise<void> }
|
||||
*/
|
||||
exports.up = async function (knex) {
|
||||
const schema = () => knex.schema.withSchema('ingestion');
|
||||
|
||||
await knex.raw(
|
||||
`CREATE INDEX IF NOT EXISTS increment_ingestion_provider_name_idx ON public.final_entities ((final_entity::json #>> '{metadata, annotations, backstage.io/incremental-provider-name}'));`,
|
||||
);
|
||||
|
||||
await knex.raw(`DROP VIEW IF EXISTS ingestion.current_entities`);
|
||||
|
||||
await schema().alterTable('ingestions', t => {
|
||||
t.primary('id');
|
||||
t.index('provider_name', 'ingestion_provider_name_idx');
|
||||
});
|
||||
|
||||
await schema().alterTable('ingestion_marks', t => {
|
||||
t.primary('id');
|
||||
t.index('ingestion_id', 'ingestion_mark_ingestion_id_idx');
|
||||
});
|
||||
|
||||
await 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
|
||||
* @returns { Promise<void> }
|
||||
*/
|
||||
exports.down = async function (knex) {
|
||||
const schema = () => knex.schema.withSchema('ingestion');
|
||||
|
||||
await schema().alterTable('ingestions', t => {
|
||||
t.dropIndex('provider_name', 'ingestion_provider_name_idx');
|
||||
t.dropPrimary('id');
|
||||
});
|
||||
|
||||
await schema().alterTable('ingestion_marks', t => {
|
||||
t.dropIndex('ingestion_id', 'ingestion_mark_ingestion_id_idx');
|
||||
t.dropPrimary('id');
|
||||
});
|
||||
|
||||
await schema().alterTable('ingestions_mark_entities', t => {
|
||||
t.dropIndex('ingestion_mark_id', 'ingestion_mark_entity_ingestion_mark_id_idx');
|
||||
t.dropPrimary('id');
|
||||
});
|
||||
|
||||
await knex.raw(`DROP INDEX increment_ingestion_provider_name_idx;`);
|
||||
};
|
||||
@@ -0,0 +1,47 @@
|
||||
{
|
||||
"name": "@backstage/plugin-incremental-ingestion-backend",
|
||||
"version": "0.1.0",
|
||||
"main": "src/index.ts",
|
||||
"types": "src/index.ts",
|
||||
"license": "Apache-2.0",
|
||||
"private": true,
|
||||
"publishConfig": {
|
||||
"access": "public",
|
||||
"main": "dist/index.cjs.js",
|
||||
"types": "dist/index.d.ts"
|
||||
},
|
||||
"backstage": {
|
||||
"role": "backend-plugin"
|
||||
},
|
||||
"scripts": {
|
||||
"start": "backstage-cli package start",
|
||||
"build": "backstage-cli package 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": "^0.14.0",
|
||||
"@backstage/backend-tasks": "^0.3.6",
|
||||
"@backstage/catalog-model": "^1.1.2",
|
||||
"@backstage/config": "^1.0.0",
|
||||
"@backstage/plugin-catalog-backend": "^1.5.0",
|
||||
"@backstage/plugin-permission-common": "^0.7.0",
|
||||
"express": "^4.17.1",
|
||||
"express-promise-router": "^4.1.0",
|
||||
"winston": "^3.2.1"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@backstage/cli": "^0.20.0",
|
||||
"@types/express": "*",
|
||||
"@types/supertest": "^2.0.8",
|
||||
"msw": "^0.35.0",
|
||||
"supertest": "^4.0.2"
|
||||
},
|
||||
"files": [
|
||||
"dist",
|
||||
"migrations/**/*.{js,d.ts}"
|
||||
]
|
||||
}
|
||||
+540
@@ -0,0 +1,540 @@
|
||||
/*
|
||||
* 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,
|
||||
IngestionRecord,
|
||||
IngestionRecordInsert,
|
||||
IngestionRecordUpdate,
|
||||
IngestionUpsertIFace,
|
||||
MarkRecord,
|
||||
MarkRecordInsert,
|
||||
} from '../types';
|
||||
|
||||
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('ingestion.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<IngestionUpsertIFace>) {
|
||||
await this.client.transaction(async tx => {
|
||||
await tx('ingestion.ingestions')
|
||||
.where('provider_name', provider)
|
||||
.andWhere('rest_completed_at', null)
|
||||
.update(update);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Performs an insert into the `ingestion.ingestions` table with the supplied values.
|
||||
* @param options IngestionRecordInsert
|
||||
*/
|
||||
async insertIngestionRecord(options: IngestionRecordInsert) {
|
||||
await this.client.transaction(async tx => {
|
||||
const { record } = options;
|
||||
await tx('ingestion.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.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>('ingestion.ingestions')
|
||||
.where('provider_name', provider)
|
||||
.andWhere('rest_completed_at', null)
|
||||
.first();
|
||||
return record;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Removes all entries from `ingestion_marks_entities`, `ingestion_marks`, and `ingestions`
|
||||
* for prior ingestions that completed (i.e., have a value for `rest_completed_at`).
|
||||
* @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.ingestion_mark_entities')
|
||||
.delete()
|
||||
.whereIn(
|
||||
'ingestion_mark_id',
|
||||
tx('ingestion.ingestion_marks')
|
||||
.select('id')
|
||||
.whereIn(
|
||||
'ingestion_id',
|
||||
tx('ingestion.ingestions')
|
||||
.select('id')
|
||||
.where('provider_name', provider)
|
||||
.andWhereNot('rest_completed_at', null),
|
||||
),
|
||||
);
|
||||
|
||||
const marksDeleted = await tx('ingestion.ingestion_marks')
|
||||
.delete()
|
||||
.whereIn(
|
||||
'ingestion_id',
|
||||
tx('ingestion.ingestions')
|
||||
.select('id')
|
||||
.where('provider_name', provider)
|
||||
.andWhereNot('rest_completed_at', null),
|
||||
);
|
||||
|
||||
const ingestionsDeleted = await tx('ingestion.ingestions')
|
||||
.delete()
|
||||
.where('provider_name', provider)
|
||||
.andWhereNot('rest_completed_at', null);
|
||||
|
||||
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>('ingestion.ingestions')
|
||||
.where('provider_name', provider)
|
||||
.andWhere('rest_completed_at', null)
|
||||
.andWhereNot('id', ingestionId);
|
||||
|
||||
if (invalid.length > 0) {
|
||||
await tx('ingestion.ingestions').delete().whereIn('id', invalid);
|
||||
await tx('ingestion.ingestion_mark_entities')
|
||||
.delete()
|
||||
.whereIn('ingestion_mark_id', tx('ingestion.ingestion_marks').select('id').whereIn('ingestion_id', invalid));
|
||||
await tx('ingestion.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('ingestion.ingestions')
|
||||
.select('id')
|
||||
.where('provider_name', provider);
|
||||
|
||||
const markIDs: { id: string }[] =
|
||||
ingestionIDs.length > 0
|
||||
? await tx('ingestion.ingestion_marks')
|
||||
.select('id')
|
||||
.whereIn(
|
||||
'ingestion_id',
|
||||
ingestionIDs.map(entry => entry.id),
|
||||
)
|
||||
: [];
|
||||
|
||||
const markEntityIDs: { id: string }[] =
|
||||
markIDs.length > 0
|
||||
? await tx('ingestion.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.ingestion_marks')
|
||||
.delete()
|
||||
.whereIn(
|
||||
'ingestion_id',
|
||||
ingestionIDs.map(entry => entry.id),
|
||||
)
|
||||
: 0;
|
||||
|
||||
const ingestionsDeleted = await tx('ingestion.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.insertRecord({
|
||||
record: {
|
||||
id: v4(),
|
||||
next_action: 'rest',
|
||||
provider_name: provider,
|
||||
next_action_at,
|
||||
ingestion_completed_at: new Date(),
|
||||
status: 'resting',
|
||||
},
|
||||
});
|
||||
|
||||
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';
|
||||
await this.insertIngestionRecord({
|
||||
record: {
|
||||
id: ingestionId,
|
||||
next_action: nextAction,
|
||||
provider_name: provider,
|
||||
status: 'bursting',
|
||||
},
|
||||
});
|
||||
return { ingestionId, nextAction, attempts: 0, nextActionAt: Date.now() };
|
||||
}
|
||||
|
||||
/**
|
||||
* 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')
|
||||
.whereRaw(
|
||||
`((final_entity::json #>> '{metadata, annotations, ${INCREMENTAL_ENTITY_PROVIDER_ANNOTATION}}')) = ?`,
|
||||
[provider],
|
||||
)
|
||||
.whereNotIn(
|
||||
'entity_ref',
|
||||
tx('ingestion.ingestion_marks')
|
||||
.join(
|
||||
'ingestion.ingestion_mark_entities',
|
||||
'ingestion.ingestion_marks.id',
|
||||
'ingestion.ingestion_mark_entities.ingestion_mark_id',
|
||||
)
|
||||
.select('ingestion.ingestion_mark_entities.ref')
|
||||
.where('ingestion.ingestion_marks.ingestion_id', ingestionId),
|
||||
);
|
||||
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 }>('ingestion.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:
|
||||
* * `ingestion.ingestions`
|
||||
* * `ingestion.ingestion_marks`
|
||||
* * `ingestion.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('ingestion.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({
|
||||
record: {
|
||||
id: v4(),
|
||||
next_action: 'rest',
|
||||
provider_name: provider,
|
||||
next_action_at,
|
||||
ingestion_completed_at: new Date(),
|
||||
status: 'resting',
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
const ingestionMarksDeleted = await this.purgeTable('ingestion.ingestion_marks');
|
||||
const markEntitiesDeleted = await this.purgeTable('ingestion.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',
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<IngestionUpsertIFace> = {
|
||||
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',
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.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.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.ingestion_marks')
|
||||
.where('ingestion_id', ingestionId)
|
||||
.orderBy('sequence', 'desc');
|
||||
return marks;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Performs an insert into the `ingestion.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.ingestion_marks').insert(record);
|
||||
});
|
||||
}
|
||||
/**
|
||||
* Performs an upsert to the `ingestion.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.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 }>('ingestion.ingestions').distinct('provider_name');
|
||||
return providers.map(entry => entry.provider_name);
|
||||
});
|
||||
}
|
||||
|
||||
async updateByName(provider: string, update: Partial<IngestionUpsertIFace>) {
|
||||
await this.updateIngestionRecordByProvider(provider, update);
|
||||
}
|
||||
async insertRecord(options: IngestionRecordInsert) {
|
||||
await this.insertIngestionRecord(options);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
/*
|
||||
* 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';
|
||||
|
||||
export async function applyDatabaseMigrations(knex: Knex): Promise<void> {
|
||||
const migrationsDir = resolvePackagePath('@devex/backend-incremental-ingestion', 'migrations');
|
||||
|
||||
await knex.raw('CREATE SCHEMA IF NOT EXISTS ingestion;');
|
||||
|
||||
await knex.migrate.latest({
|
||||
schemaName: 'ingestion',
|
||||
directory: migrationsDir,
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,218 @@
|
||||
/*
|
||||
* 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';
|
||||
|
||||
export class IncrementalIngestionEngine implements IterationEngine {
|
||||
restLength: Duration;
|
||||
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 { ingestionId, nextActionAt, nextAction, attempts } = await this.getCurrentAction();
|
||||
|
||||
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 = (error as string).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}'`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
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,
|
||||
};
|
||||
} else {
|
||||
const result = await this.manager.createProviderIngestionRecord(providerName);
|
||||
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 : void 0;
|
||||
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++;
|
||||
while (true) {
|
||||
done = next.done;
|
||||
await this.mark(id, sequence, next.entities, next.done, 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(id: string, sequence: number, entities: DeferredEntity[], done: boolean, cursor?: unknown) {
|
||||
this.options.logger.debug(
|
||||
`incremental-engine: Ingestion '${id}': MARK ${entities.length} entities, cursor: ${JSON.stringify(
|
||||
cursor,
|
||||
)}, done: ${done}`,
|
||||
);
|
||||
const markId = v4();
|
||||
|
||||
await this.manager.createMark({
|
||||
record: {
|
||||
id: markId,
|
||||
ingestion_id: id,
|
||||
cursor,
|
||||
sequence,
|
||||
},
|
||||
});
|
||||
|
||||
if (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[] = done
|
||||
? []
|
||||
: await this.manager.computeRemoved(this.options.provider.getProviderName(), id);
|
||||
|
||||
await this.options.connection.applyMutation({
|
||||
type: 'delta',
|
||||
added,
|
||||
removed,
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
/*
|
||||
* 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 * from './service/IncrementalCatalogBuilder';
|
||||
export * from './types';
|
||||
export * from './database/IncrementalIngestionDatabaseManager';
|
||||
@@ -0,0 +1,202 @@
|
||||
/*
|
||||
* 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';
|
||||
|
||||
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('/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.delete('/cleanup', async (_, res) => {
|
||||
const result = await manager.cleanupProviders();
|
||||
res.json(result);
|
||||
});
|
||||
|
||||
// Get basic status of the provider
|
||||
router.get('/:provider', 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.put('/:provider', 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
|
||||
router.post('/:provider', async (req, res) => {
|
||||
const { provider } = req.params;
|
||||
|
||||
const record = await manager.getCurrentIngestionRecord(provider);
|
||||
if (record) {
|
||||
await manager.updateByName(provider, {
|
||||
next_action: 'nothing (done)',
|
||||
ingestion_completed_at: new Date(),
|
||||
rest_completed_at: new Date(),
|
||||
status: 'complete',
|
||||
});
|
||||
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/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', 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/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/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,127 @@
|
||||
/*
|
||||
* 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 '../routes';
|
||||
|
||||
export class Deferred<T> implements Promise<T> {
|
||||
// eslint-disable-next-line @typescript-eslint/ban-ts-comment
|
||||
//@ts-ignore
|
||||
resolve: (value: T) => void;
|
||||
// eslint-disable-next-line @typescript-eslint/ban-ts-comment
|
||||
//@ts-ignore
|
||||
reject: (error: Error) => void;
|
||||
|
||||
then: Promise<T>['then'];
|
||||
catch: Promise<T>['catch'];
|
||||
finally: Promise<T>['finally'];
|
||||
|
||||
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';
|
||||
}
|
||||
|
||||
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, manager: this.manager };
|
||||
}
|
||||
|
||||
addIncrementalEntityProvider<T, C>(
|
||||
provider: IncrementalEntityProvider<T, C>,
|
||||
options: IncrementalEntityProviderOptions,
|
||||
) {
|
||||
const { burstInterval, burstLength, restLength } = options;
|
||||
const { logger: catalogLogger, database, scheduler } = this.env;
|
||||
const ready = this.ready;
|
||||
|
||||
this.builder.addEntityProvider({
|
||||
getProviderName: provider.getProviderName.bind(provider),
|
||||
async connect(connection) {
|
||||
const logger = catalogLogger.child({ entityProvider: provider.getProviderName() });
|
||||
|
||||
logger.info(`Connecting`);
|
||||
|
||||
const client = await database.getClient();
|
||||
|
||||
const manager = new IncrementalIngestionDatabaseManager({ client });
|
||||
|
||||
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,221 @@
|
||||
/*
|
||||
* 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 { PermissionAuthorizer } 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.
|
||||
*
|
||||
*/
|
||||
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 uniqiue 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>>;
|
||||
|
||||
/**
|
||||
* 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.
|
||||
*/
|
||||
export interface EntityIteratorResult<T> {
|
||||
/**
|
||||
* Indicates whether there are any further pages of entities to
|
||||
* ingest after this one.
|
||||
*/
|
||||
done: boolean;
|
||||
|
||||
/**
|
||||
* A value that marks the page of entities after this one. It will
|
||||
* be used to pass into the following invocation of `next()`
|
||||
*/
|
||||
cursor: T;
|
||||
|
||||
/**
|
||||
* The entities to ingest.
|
||||
*/
|
||||
entities: DeferredEntity[];
|
||||
}
|
||||
|
||||
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[];
|
||||
}
|
||||
|
||||
export type PluginEnvironment = {
|
||||
logger: Logger;
|
||||
database: PluginDatabaseManager;
|
||||
scheduler: PluginTaskScheduler;
|
||||
config: Config;
|
||||
reader: UrlReader;
|
||||
permissions: PermissionAuthorizer;
|
||||
};
|
||||
|
||||
/**
|
||||
* The core ingestion engine implements this interface
|
||||
*/
|
||||
export interface IterationEngine {
|
||||
taskFn: TaskFunction;
|
||||
}
|
||||
|
||||
/**
|
||||
* Options passed to the core ingestion engine during initialization.
|
||||
*/
|
||||
export interface IterationEngineOptions {
|
||||
logger: Logger;
|
||||
connection: EntityProviderConnection;
|
||||
manager: IncrementalIngestionDatabaseManager;
|
||||
provider: IncrementalEntityProvider<unknown, unknown>;
|
||||
restLength: DurationObjectUnits;
|
||||
ready: Promise<void>;
|
||||
backoff?: IncrementalEntityProviderOptions['backoff'];
|
||||
}
|
||||
|
||||
/**
|
||||
* The shape of data inserted into or updated in the `ingestion.ingestions` table.
|
||||
*/
|
||||
export interface IngestionUpsertIFace {
|
||||
next_action: 'rest' | 'ingest' | 'backoff' | 'cancel' | 'nothing (done)' | 'nothing (canceled)';
|
||||
status: 'complete' | 'bursting' | 'resting' | 'canceling' | 'interstitial' | 'backing off';
|
||||
provider_name: string;
|
||||
next_action_at?: Date;
|
||||
last_error?: string;
|
||||
attempts?: number;
|
||||
ingestion_completed_at?: Date;
|
||||
rest_completed_at?: Date;
|
||||
}
|
||||
|
||||
/**
|
||||
* This interface supplies all potential values that can be inserted into the `ingestion.ingestions` table.
|
||||
*/
|
||||
export interface IngestionRecordInsert {
|
||||
record: IngestionUpsertIFace & {
|
||||
id: string;
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* This interface is for updating an existing ingestion record.
|
||||
*/
|
||||
export interface IngestionRecordUpdate {
|
||||
ingestionId: string;
|
||||
update: Partial<IngestionUpsertIFace>;
|
||||
}
|
||||
|
||||
/**
|
||||
* The expected response from the `ingestion.ingestion_marks` table.
|
||||
*/
|
||||
export interface MarkRecord {
|
||||
id: string;
|
||||
sequence: number;
|
||||
ingestion_id: string;
|
||||
cursor: string;
|
||||
created_at: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* The expected response from the `ingestion.ingestions` table.
|
||||
*/
|
||||
export interface IngestionRecord {
|
||||
id: string;
|
||||
provider_name: string;
|
||||
status: string;
|
||||
next_action: string;
|
||||
next_action_at: Date;
|
||||
last_error: string | null;
|
||||
attempts: number;
|
||||
created_at: string;
|
||||
rest_completed_at: string | null;
|
||||
ingestion_completed_at: string | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* This interface supplies all the values for adding an ingestion mark.
|
||||
*/
|
||||
export interface MarkRecordInsert {
|
||||
record: {
|
||||
id: string;
|
||||
ingestion_id: string;
|
||||
cursor: unknown;
|
||||
sequence: number;
|
||||
};
|
||||
}
|
||||
@@ -6023,6 +6023,25 @@ __metadata:
|
||||
languageName: unknown
|
||||
linkType: soft
|
||||
|
||||
"@backstage/plugin-incremental-ingestion-backend@^0.0.0, @backstage/plugin-incremental-ingestion-backend@workspace:plugins/incremental-ingestion-backend":
|
||||
version: 0.0.0-use.local
|
||||
resolution: "@backstage/plugin-incremental-ingestion-backend@workspace:plugins/incremental-ingestion-backend"
|
||||
dependencies:
|
||||
"@backstage/backend-common": "workspace:^"
|
||||
"@backstage/cli": "workspace:^"
|
||||
"@backstage/config": "workspace:^"
|
||||
"@types/express": "*"
|
||||
"@types/supertest": ^2.0.8
|
||||
express: ^4.18.1
|
||||
express-promise-router: ^4.1.0
|
||||
msw: ^0.47.0
|
||||
node-fetch: ^2.6.7
|
||||
supertest: ^6.2.4
|
||||
winston: ^3.2.1
|
||||
yn: ^4.0.0
|
||||
languageName: unknown
|
||||
linkType: soft
|
||||
|
||||
"@backstage/plugin-jenkins-backend@workspace:^, @backstage/plugin-jenkins-backend@workspace:plugins/jenkins-backend":
|
||||
version: 0.0.0-use.local
|
||||
resolution: "@backstage/plugin-jenkins-backend@workspace:plugins/jenkins-backend"
|
||||
@@ -21350,6 +21369,7 @@ __metadata:
|
||||
"@backstage/plugin-events-backend": "workspace:^"
|
||||
"@backstage/plugin-events-node": "workspace:^"
|
||||
"@backstage/plugin-graphql-backend": "workspace:^"
|
||||
"@backstage/plugin-incremental-ingestion-backend": ^0.0.0
|
||||
"@backstage/plugin-jenkins-backend": "workspace:^"
|
||||
"@backstage/plugin-kafka-backend": "workspace:^"
|
||||
"@backstage/plugin-kubernetes-backend": "workspace:^"
|
||||
|
||||
Reference in New Issue
Block a user