From 2ef75976b1976a170377506f6a5ebadc96721ff9 Mon Sep 17 00:00:00 2001 From: Damon Kaswell Date: Thu, 27 Oct 2022 16:25:17 -0700 Subject: [PATCH] Run prettier on sources Signed-off-by: Damon Kaswell --- .../incremental-ingestion-backend/README.md | 87 ++++++++------- ...21019897897_add-unique-active-ingestion.js | 28 +++-- .../IncrementalIngestionDatabaseManager.ts | 87 +++++++++++---- .../src/database/migrations.ts | 5 +- .../src/engine/IncrementalIngestionEngine.ts | 101 +++++++++++++----- .../src/routes.ts | 53 +++++++-- .../src/service/IncrementalCatalogBuilder.ts | 27 +++-- .../src/types.ts | 41 +++++-- 8 files changed, 318 insertions(+), 111 deletions(-) diff --git a/plugins/incremental-ingestion-backend/README.md b/plugins/incremental-ingestion-backend/README.md index d06334f396..4e9d4fe65b 100644 --- a/plugins/incremental-ingestion-backend/README.md +++ b/plugins/incremental-ingestion-backend/README.md @@ -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 { - 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 { * @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>; + next( + context: TContext, + cursor?: TCursor, + ): Promise>; } ``` @@ -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 + getServices(page: number): MyPaginatedResults; } interface MyPaginatedResults { @@ -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 { +export class MyIncrementalEntityProvider + implements IncrementalEntityProvider +{ getProviderName() { return `MyIncrementalEntityProvider`; } @@ -175,13 +186,14 @@ export class MyIncrementalEntityProvider implements IncrementalEntityProvider { +export class MyIncrementalEntityProvider + implements IncrementalEntityProvider +{ getProviderName() { return `MyIncrementalEntityProvider`; } async around(burst: (context: MyContext) => Promise): Promise { - const apiClient = new MyApiClient(); await burst({ apiClient }); @@ -194,10 +206,11 @@ export class MyIncrementalEntityProvider implements IncrementalEntityProvider { - +export class MyIncrementalEntityProvider + implements IncrementalEntityProvider +{ token: string; - + constructor(token: string) { this.token = token; } @@ -206,12 +219,10 @@ export class MyIncrementalEntityProvider implements IncrementalEntityProvider Promise): Promise { + 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 { - + token: string; - + constructor(token: string) { this.token = token; } @@ -284,7 +295,7 @@ export class MyIncrementalEntityProvider implements IncrementalEntityProvider 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); } }); diff --git a/plugins/incremental-ingestion-backend/src/database/IncrementalIngestionDatabaseManager.ts b/plugins/incremental-ingestion-backend/src/database/IncrementalIngestionDatabaseManager.ts index b736693df8..d593bf4292 100644 --- a/plugins/incremental-ingestion-backend/src/database/IncrementalIngestionDatabaseManager.ts +++ b/plugins/incremental-ingestion-backend/src/database/IncrementalIngestionDatabaseManager.ts @@ -51,7 +51,10 @@ export class IncrementalIngestionDatabaseManager { * @param provider string * @param update Partial */ - async updateIngestionRecordByProvider(provider: string, update: Partial) { + async updateIngestionRecordByProvider( + provider: string, + update: Partial, + ) { 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); }); } diff --git a/plugins/incremental-ingestion-backend/src/database/migrations.ts b/plugins/incremental-ingestion-backend/src/database/migrations.ts index 8ec5d3a7e6..991a709732 100644 --- a/plugins/incremental-ingestion-backend/src/database/migrations.ts +++ b/plugins/incremental-ingestion-backend/src/database/migrations.ts @@ -17,7 +17,10 @@ import { resolvePackagePath } from '@backstage/backend-common'; import { Knex } from 'knex'; export async function applyDatabaseMigrations(knex: Knex): Promise { - 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;'); diff --git a/plugins/incremental-ingestion-backend/src/engine/IncrementalIngestionEngine.ts b/plugins/incremental-ingestion-backend/src/engine/IncrementalIngestionEngine.ts index bc347310a8..655380cec4 100644 --- a/plugins/incremental-ingestion-backend/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/incremental-ingestion-backend/src/engine/IncrementalIngestionEngine.ts @@ -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', diff --git a/plugins/incremental-ingestion-backend/src/routes.ts b/plugins/incremental-ingestion-backend/src/routes.ts index 8e2f3ae8af..ca1743113e 100644 --- a/plugins/incremental-ingestion-backend/src/routes.ts +++ b/plugins/incremental-ingestion-backend/src/routes.ts @@ -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: {}, diff --git a/plugins/incremental-ingestion-backend/src/service/IncrementalCatalogBuilder.ts b/plugins/incremental-ingestion-backend/src/service/IncrementalCatalogBuilder.ts index 30bdae9746..a9babf17e4 100644 --- a/plugins/incremental-ingestion-backend/src/service/IncrementalCatalogBuilder.ts +++ b/plugins/incremental-ingestion-backend/src/service/IncrementalCatalogBuilder.ts @@ -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(), diff --git a/plugins/incremental-ingestion-backend/src/types.ts b/plugins/incremental-ingestion-backend/src/types.ts index 4a63391273..85f184b52e 100644 --- a/plugins/incremental-ingestion-backend/src/types.ts +++ b/plugins/incremental-ingestion-backend/src/types.ts @@ -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 { * @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>; + next( + context: TContext, + cursor?: TCursor, + ): Promise>; /** * 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;