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

Run prettier on sources

Browse Source 
Signed-off-by: Damon Kaswell <damon.kaswell1@hp.com>
This commit is contained in:
Damon Kaswell
2022-10-27 16:25:17 -07:00
committed by Fredrik Adelöw
parent 2389b4d521
commit 2ef75976b1
8 changed files with 318 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/incremental-ingestion-backend/README.md
+49 -38
View File
@@ -44,27 +44,28 @@ The Incremental Entity Provider backend is designed for data sources that provid
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();
```
```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 { CatalogBuilder } from '@backstage/plugin-catalog-backend';
import { ScaffolderEntitiesProcessor } from '@backstage/plugin-scaffolder-backend';
import { IncrementalCatalogBuilder } from '@backstage/plugin-incremental-ingestion-backend';
import { Router } from 'express';
@@ -74,18 +75,20 @@ 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);
// 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
// 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);
@@ -128,7 +131,10 @@ interface IncrementalEntityProvider<TCursor, TContext> {
* @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>>;
next(
context: TContext,
cursor?: TCursor,
): Promise<EntityIteratorResult<TCursor>>;
}
```
@@ -136,7 +142,7 @@ For tutorial, we'll write an Incremental Entity Provider that will call an imagi
```ts
interface MyApiClient {
getServices(page: number): MyPaginatedResults<Service>
getServices(page: number): MyPaginatedResults<Service>;
}
interface MyPaginatedResults<T> {
@@ -152,7 +158,10 @@ interface Service {
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';
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`
@@ -162,10 +171,12 @@ interface MyApiCursor {
// This interface describes the type of data that will be passed to your burst function.
interface MyContext {
apiClient: MyApiClient
apiClient: MyApiClient;
}
export class MyIncrementalEntityProvider implements IncrementalEntityProvider<MyApiCursor, MyContext> {
export class MyIncrementalEntityProvider
implements IncrementalEntityProvider<MyApiCursor, MyContext>
{
getProviderName() {
return `MyIncrementalEntityProvider`;
}
@@ -175,13 +186,14 @@ export class MyIncrementalEntityProvider implements IncrementalEntityProvider<My
`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> {
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 });
@@ -194,10 +206,11 @@ export class MyIncrementalEntityProvider implements IncrementalEntityProvider<Cu
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> {
export class MyIncrementalEntityProvider
implements IncrementalEntityProvider<Cursor, Context>
{
token: string;
constructor(token: string) {
this.token = token;
}
@@ -206,12 +219,10 @@ export class MyIncrementalEntityProvider implements IncrementalEntityProvider<Cu
return `MyIncrementalEntityProvider`;
}
async around(burst: (context: MyContext) => Promise<void>): Promise<void> {
const apiClient = new MyApiClient(this.token);
const apiClient = new MyApiClient(this.token)
await burst({ apiClient })
await burst({ apiClient });
}
}
```
@@ -220,9 +231,9 @@ The last step is to implement the actual `next` method that will accept the curs
```ts
export class MyIncrementalEntityProvider implements IncrementalEntityProvider<Cursor, Context> {
token: string;
constructor(token: string) {
this.token = token;
}
@@ -284,7 +295,7 @@ export class MyIncrementalEntityProvider implements IncrementalEntityProvider<Cu
}
```
Now that you have your new Incremental Entity Provider, we can connect it to the catalog.
Now that you have your new Incremental Entity Provider, we can connect it to the catalog.
## Adding an Incremental Entity Provider to the catalog
plugins/incremental-ingestion-backend/migrations/20221019897897_add-unique-active-ingestion.js
+22 -6
View File
@@ -23,7 +23,9 @@ exports.up = async function up(knex) {
const schema = () => knex.schema.withSchema('ingestion');
await knex.transaction(async tx => {
const providers = await tx('ingestion.ingestions').distinct('provider_name');
const providers = await tx('ingestion.ingestions').distinct(
'provider_name',
);
for (const provider of providers) {
const valid = await tx('ingestion.ingestions')
.where('provider_name', provider)
@@ -44,8 +46,15 @@ exports.up = async function up(knex) {
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);
.whereIn(
'ingestion_mark_id',
tx('ingestion.ingestion_marks')
.select('id')
.whereIn('ingestion_id', invalid),
);
await tx('ingestion.ingestion_marks')
.delete()
.whereIn('ingestion_id', invalid);
}
}
});
@@ -55,11 +64,18 @@ exports.up = async function up(knex) {
});
await knex.transaction(async tx => {
await tx('ingestion.ingestions').update('completion_ticket', 'open').where('rest_completed_at', null);
await tx('ingestion.ingestions')
.update('completion_ticket', 'open')
.where('rest_completed_at', null);
const rows = await tx('ingestion.ingestions').whereNot('rest_completed_at', null);
const rows = await tx('ingestion.ingestions').whereNot(
'rest_completed_at',
null,
);
for (const row of rows) {
await tx('ingestion.ingestions').update('completion_ticket', uuidv4()).where('id', row.id);
await tx('ingestion.ingestions')
.update('completion_ticket', uuidv4())
.where('id', row.id);
}
});
plugins/incremental-ingestion-backend/src/database/IncrementalIngestionDatabaseManager.ts
+69 -18
View File
@@ -51,7 +51,10 @@ export class IncrementalIngestionDatabaseManager {
* @param provider string
* @param update Partial<IngestionUpsertIFace>
*/
async updateIngestionRecordByProvider(provider: string, update: Partial<IngestionUpsertIFace>) {
async updateIngestionRecordByProvider(
provider: string,
update: Partial<IngestionUpsertIFace>,
) {
await this.client.transaction(async tx => {
await tx('ingestion.ingestions')
.where('provider_name', provider)
@@ -71,7 +74,10 @@ export class IncrementalIngestionDatabaseManager {
});
}
private async deleteMarkEntities(tx: Knex.Transaction, ids: { id: string }[]) {
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);
@@ -174,8 +180,15 @@ export class IncrementalIngestionDatabaseManager {
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);
.whereIn(
'ingestion_mark_id',
tx('ingestion.ingestion_marks')
.select('id')
.whereIn('ingestion_id', invalid),
);
await tx('ingestion.ingestion_marks')
.delete()
.whereIn('ingestion_id', invalid);
}
});
}
@@ -212,7 +225,10 @@ export class IncrementalIngestionDatabaseManager {
)
: [];
const markEntitiesDeleted = await this.deleteMarkEntities(tx, markEntityIDs);
const markEntitiesDeleted = await this.deleteMarkEntities(
tx,
markEntityIDs,
);
const marksDeleted =
markIDs.length > 0
@@ -224,7 +240,9 @@ export class IncrementalIngestionDatabaseManager {
)
: 0;
const ingestionsDeleted = await tx('ingestion.ingestions').delete().where('provider_name', provider);
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);
@@ -271,9 +289,18 @@ export class IncrementalIngestionDatabaseManager {
*/
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')
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],
@@ -301,7 +328,9 @@ export class IncrementalIngestionDatabaseManager {
*/
async healthcheck() {
return await this.client.transaction(async tx => {
const records = await tx<{ id: string; provider_name: string }>('ingestion.ingestions')
const records = await tx<{ id: string; provider_name: string }>(
'ingestion.ingestions',
)
.distinct('id', 'provider_name')
.where('rest_completed_at', null);
return records;
@@ -313,7 +342,9 @@ export class IncrementalIngestionDatabaseManager {
* @param provider string
*/
async triggerNextProviderAction(provider: string) {
await this.updateIngestionRecordByProvider(provider, { next_action_at: new Date() });
await this.updateIngestionRecordByProvider(provider, {
next_action_at: new Date(),
});
}
/**
@@ -346,8 +377,12 @@ export class IncrementalIngestionDatabaseManager {
});
}
const ingestionMarksDeleted = await this.purgeTable('ingestion.ingestion_marks');
const markEntitiesDeleted = await this.purgeTable('ingestion.ingestion_mark_entities');
const ingestionMarksDeleted = await this.purgeTable(
'ingestion.ingestion_marks',
);
const markEntitiesDeleted = await this.purgeTable(
'ingestion.ingestion_mark_entities',
);
return { ingestionsDeleted, ingestionMarksDeleted, markEntitiesDeleted };
}
@@ -357,7 +392,10 @@ export class IncrementalIngestionDatabaseManager {
* @param ingestionId string
*/
async setProviderIngesting(ingestionId: string) {
await this.updateIngestionRecordById({ ingestionId, update: { next_action: 'ingest' } });
await this.updateIngestionRecordById({
ingestionId,
update: { next_action: 'ingest' },
});
}
/**
@@ -365,7 +403,10 @@ export class IncrementalIngestionDatabaseManager {
* @param ingestionId string
*/
async setProviderBursting(ingestionId: string) {
await this.updateIngestionRecordById({ ingestionId, update: { status: 'bursting' } });
await this.updateIngestionRecordById({
ingestionId,
update: { status: 'bursting' },
});
}
/**
@@ -405,7 +446,10 @@ export class IncrementalIngestionDatabaseManager {
* @param ingestionId string
*/
async setProviderInterstitial(ingestionId: string) {
await this.updateIngestionRecordById({ ingestionId, update: { attempts: 0, status: 'interstitial' } });
await this.updateIngestionRecordById({
ingestionId,
update: { attempts: 0, status: 'interstitial' },
});
}
/**
@@ -445,7 +489,12 @@ export class IncrementalIngestionDatabaseManager {
* @param error Error
* @param backoffLength number
*/
async setProviderBackoff(ingestionId: string, attempts: number, error: Error, backoffLength: number) {
async setProviderBackoff(
ingestionId: string,
attempts: number,
error: Error,
backoffLength: number,
) {
await this.updateIngestionRecordById({
ingestionId,
update: {
@@ -526,7 +575,9 @@ export class IncrementalIngestionDatabaseManager {
*/
async listProviders() {
return await this.client.transaction(async tx => {
const providers = await tx<{ provider_name: string }>('ingestion.ingestions').distinct('provider_name');
const providers = await tx<{ provider_name: string }>(
'ingestion.ingestions',
).distinct('provider_name');
return providers.map(entry => entry.provider_name);
});
}
plugins/incremental-ingestion-backend/src/database/migrations.ts
+4 -1
View File
@@ -17,7 +17,10 @@ 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');
const migrationsDir = resolvePackagePath(
'@devex/backend-incremental-ingestion',
'migrations',
);
await knex.raw('CREATE SCHEMA IF NOT EXISTS ingestion;');
plugins/incremental-ingestion-backend/src/engine/IncrementalIngestionEngine.ts
+77 -24
View File
@@ -14,7 +14,11 @@
* limitations under the License.
*/
import type { DeferredEntity } from '@backstage/plugin-catalog-backend';
import { INCREMENTAL_ENTITY_PROVIDER_ANNOTATION, IterationEngine, IterationEngineOptions } from '../types';
import {
INCREMENTAL_ENTITY_PROVIDER_ANNOTATION,
IterationEngine,
IterationEngineOptions,
} from '../types';
import { IncrementalIngestionDatabaseManager } from '../database/IncrementalIngestionDatabaseManager';
import type { AbortSignal } from 'node-abort-controller';
@@ -31,7 +35,12 @@ export class IncrementalIngestionEngine implements IterationEngine {
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 }];
this.backoff = options.backoff ?? [
{ minutes: 1 },
{ minutes: 5 },
{ minutes: 30 },
{ hours: 3 },
];
}
async taskFn(signal: AbortSignal) {
@@ -49,19 +58,24 @@ export class IncrementalIngestionEngine implements IterationEngine {
async handleNextAction(signal: AbortSignal) {
await this.options.ready;
const { ingestionId, nextActionAt, nextAction, attempts } = await this.getCurrentAction();
const { ingestionId, nextActionAt, nextAction, attempts } =
await this.getCurrentAction();
switch (nextAction) {
case 'rest':
if (Date.now() > nextActionAt) {
await this.manager.clearFinishedIngestions(this.options.provider.getProviderName());
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`);
this.options.logger.info(
`incremental-engine: Ingestion '${ingestionId}' rest period continuing`,
);
}
break;
case 'ingest':
@@ -75,14 +89,26 @@ export class IncrementalIngestionEngine implements IterationEngine {
await this.manager.setProviderResting(ingestionId, this.restLength);
} else {
await this.manager.setProviderInterstitial(ingestionId);
this.options.logger.info(`incremental-engine: Ingestion '${ingestionId}' continuing`);
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);
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 currentBackoff = Duration.fromObject(
this.backoff[Math.min(this.backoff.length - 1, attempts)],
);
const backoffLength = currentBackoff.as('milliseconds');
this.options.logger.error(error);
@@ -92,7 +118,12 @@ export class IncrementalIngestionEngine implements IterationEngine {
`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);
await this.manager.setProviderBackoff(
ingestionId,
attempts,
error as Error,
backoffLength,
);
}
}
break;
@@ -103,11 +134,15 @@ export class IncrementalIngestionEngine implements IterationEngine {
);
await this.manager.setProviderIngesting(ingestionId);
} else {
this.options.logger.info(`incremental-engine: Ingestion '${ingestionId}' backoff continuing`);
this.options.logger.info(
`incremental-engine: Ingestion '${ingestionId}' backoff continuing`,
);
}
break;
case 'cancel':
this.options.logger.info(`incremental-engine: Ingestion '${ingestionId}' canceling, will restart`);
this.options.logger.info(
`incremental-engine: Ingestion '${ingestionId}' canceling, will restart`,
);
await this.manager.setProviderCanceled(ingestionId);
break;
default:
@@ -121,16 +156,22 @@ export class IncrementalIngestionEngine implements IterationEngine {
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}'`);
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);
this.options.logger.info(`incremental-engine: Ingestion record created: '${result.ingestionId}'`);
}
const result = await this.manager.createProviderIngestionRecord(
providerName,
);
this.options.logger.info(
`incremental-engine: Ingestion record created: '${result.ingestionId}'`,
);
return result;
}
@@ -143,7 +184,9 @@ export class IncrementalIngestionEngine implements IterationEngine {
const start = performance.now();
let count = 0;
let done = false;
this.options.logger.info(`incremental-engine: Ingestion '${id}' burst initiated`);
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);
@@ -170,11 +213,17 @@ export class IncrementalIngestionEngine implements IterationEngine {
return done;
}
async mark(id: string, sequence: number, entities: DeferredEntity[], done: boolean, cursor?: unknown) {
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}`,
`incremental-engine: Ingestion '${id}': MARK ${
entities.length
} entities, cursor: ${JSON.stringify(cursor)}, done: ${done}`,
);
const markId = v4();
@@ -199,7 +248,8 @@ export class IncrementalIngestionEngine implements IterationEngine {
...deferred.entity.metadata,
annotations: {
...deferred.entity.metadata.annotations,
[INCREMENTAL_ENTITY_PROVIDER_ANNOTATION]: this.options.provider.getProviderName(),
[INCREMENTAL_ENTITY_PROVIDER_ANNOTATION]:
this.options.provider.getProviderName(),
},
},
},
@@ -207,7 +257,10 @@ export class IncrementalIngestionEngine implements IterationEngine {
const removed: DeferredEntity[] = done
? []
: await this.manager.computeRemoved(this.options.provider.getProviderName(), id);
: await this.manager.computeRemoved(
this.options.provider.getProviderName(),
id,
);
await this.options.connection.applyMutation({
type: 'delta',
plugins/incremental-ingestion-backend/src/routes.ts
+43 -10
View File
@@ -19,7 +19,10 @@ import Router from 'express-promise-router';
import { Logger } from 'winston';
import { IncrementalIngestionDatabaseManager } from './database/IncrementalIngestionDatabaseManager';
export const createIncrementalProviderRouter = async (manager: IncrementalIngestionDatabaseManager, logger: Logger) => {
export const createIncrementalProviderRouter = async (
manager: IncrementalIngestionDatabaseManager,
logger: Logger,
) => {
const router = Router();
router.use(express.json());
@@ -27,7 +30,9 @@ export const createIncrementalProviderRouter = async (manager: IncrementalIngest
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))];
const duplicates = [
...new Set(providers.filter((e, i, a) => a.indexOf(e) !== i)),
];
if (duplicates.length > 0) {
res.json({ healthy: false, duplicateIngestions: duplicates });
@@ -65,7 +70,9 @@ export const createIncrementalProviderRouter = async (manager: IncrementalIngest
},
});
} else {
logger.error(`${provider} - No ingestion record found in the database!`);
logger.error(
`${provider} - No ingestion record found in the database!`,
);
res.status(404).json({
success: false,
status: {},
@@ -81,7 +88,10 @@ export const createIncrementalProviderRouter = async (manager: IncrementalIngest
const record = await manager.getCurrentIngestionRecord(provider);
if (record) {
await manager.triggerNextProviderAction(provider);
res.json({ success: true, message: `${provider}: Next action triggered.` });
res.json({
success: true,
message: `${provider}: Next action triggered.`,
});
} else {
const providers: string[] = await manager.listProviders();
if (providers.includes(provider)) {
@@ -91,7 +101,12 @@ export const createIncrementalProviderRouter = async (manager: IncrementalIngest
message: 'Unable to trigger next action (provider is restarting)',
});
} else {
res.status(404).json({ success: false, message: `Provider '${provider}' not found` });
res
.status(404)
.json({
success: false,
message: `Provider '${provider}' not found`,
});
}
}
});
@@ -108,7 +123,10 @@ export const createIncrementalProviderRouter = async (manager: IncrementalIngest
rest_completed_at: new Date(),
status: 'complete',
});
res.json({ success: true, message: `${provider}: Next cycle triggered.` });
res.json({
success: true,
message: `${provider}: Next cycle triggered.`,
});
} else {
const providers: string[] = await manager.listProviders();
if (providers.includes(provider)) {
@@ -118,7 +136,12 @@ export const createIncrementalProviderRouter = async (manager: IncrementalIngest
message: 'Provider is already restarting',
});
} else {
res.status(404).json({ success: false, message: `Provider '${provider}' not found` });
res
.status(404)
.json({
success: false,
message: `Provider '${provider}' not found`,
});
}
}
});
@@ -136,7 +159,10 @@ export const createIncrementalProviderRouter = async (manager: IncrementalIngest
next_action_at,
status: 'resting',
});
res.json({ success: true, message: `${provider}: Current ingestion canceled.` });
res.json({
success: true,
message: `${provider}: Current ingestion canceled.`,
});
} else {
const providers: string[] = await manager.listProviders();
if (providers.includes(provider)) {
@@ -146,7 +172,12 @@ export const createIncrementalProviderRouter = async (manager: IncrementalIngest
message: 'Provider is currently restarting, please wait.',
});
} else {
res.status(404).json({ success: false, message: `Provider '${provider}' not found` });
res
.status(404)
.json({
success: false,
message: `Provider '${provider}' not found`,
});
}
}
});
@@ -175,7 +206,9 @@ export const createIncrementalProviderRouter = async (manager: IncrementalIngest
message: 'No records yet (provider is restarting)',
});
} else {
logger.error(`${provider} - No ingestion record found in the database!`);
logger.error(
`${provider} - No ingestion record found in the database!`,
);
res.status(404).json({
success: false,
status: {},
plugins/incremental-ingestion-backend/src/service/IncrementalCatalogBuilder.ts
+21 -6
View File
@@ -13,7 +13,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import { IncrementalEntityProvider, IncrementalEntityProviderOptions, PluginEnvironment } from '../types';
import {
IncrementalEntityProvider,
IncrementalEntityProviderOptions,
PluginEnvironment,
} from '../types';
import { CatalogBuilder as CoreCatalogBuilder } from '@backstage/plugin-catalog-backend';
import { Duration } from 'luxon';
import { Knex } from 'knex';
@@ -76,9 +80,14 @@ export class IncrementalCatalogBuilder {
await applyDatabaseMigrations(this.client);
this.ready.resolve();
const routerLogger = this.env.logger.child({ router: 'IncrementalProviderAdmin' });
const routerLogger = this.env.logger.child({
router: 'IncrementalProviderAdmin',
});
const incrementalAdminRouter = await createIncrementalProviderRouter(this.manager, routerLogger);
const incrementalAdminRouter = await createIncrementalProviderRouter(
this.manager,
routerLogger,
);
return { incrementalAdminRouter, manager: this.manager };
}
@@ -94,7 +103,9 @@ export class IncrementalCatalogBuilder {
this.builder.addEntityProvider({
getProviderName: provider.getProviderName.bind(provider),
async connect(connection) {
const logger = catalogLogger.child({ entityProvider: provider.getProviderName() });
const logger = catalogLogger.child({
entityProvider: provider.getProviderName(),
});
logger.info(`Connecting`);
@@ -112,8 +123,12 @@ export class IncrementalCatalogBuilder {
connection,
});
const frequency = Duration.isDuration(burstInterval) ? burstInterval : Duration.fromObject(burstInterval);
const length = Duration.isDuration(burstLength) ? burstLength : Duration.fromObject(burstLength);
const frequency = Duration.isDuration(burstInterval)
? burstInterval
: Duration.fromObject(burstInterval);
const length = Duration.isDuration(burstLength)
? burstLength
: Duration.fromObject(burstLength);
await scheduler.scheduleTask({
id: provider.getProviderName(),
plugins/incremental-ingestion-backend/src/types.ts
+33 -8
View File
@@ -13,10 +13,19 @@
* 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 {
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 {
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';
@@ -27,7 +36,8 @@ import { IncrementalIngestionDatabaseManager } from './database/IncrementalInges
*
* @public
*/
export const INCREMENTAL_ENTITY_PROVIDER_ANNOTATION = 'backstage.io/incremental-provider-name';
export const INCREMENTAL_ENTITY_PROVIDER_ANNOTATION =
'backstage.io/incremental-provider-name';
/**
* Ingest entities into the catalog in bite-sized chunks.
@@ -58,7 +68,10 @@ export interface IncrementalEntityProvider<TCursor, TContext> {
* @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>>;
next(
context: TContext,
cursor?: TCursor,
): Promise<EntityIteratorResult<TCursor>>;
/**
* Do any setup and teardown necessary in order to provide the
@@ -131,7 +144,7 @@ export type PluginEnvironment = {
};
/**
* The core ingestion engine implements this interface
* The core ingestion engine implements this interface
*/
export interface IterationEngine {
taskFn: TaskFunction;
@@ -154,8 +167,20 @@ export interface IterationEngineOptions {
* 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';
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;