diff --git a/plugins/incremental-ingestion-backend/src/database/IncrementalIngestionDatabaseManager.ts b/plugins/incremental-ingestion-backend/src/database/IncrementalIngestionDatabaseManager.ts index 6a59757ca1..cc791a6f88 100644 --- a/plugins/incremental-ingestion-backend/src/database/IncrementalIngestionDatabaseManager.ts +++ b/plugins/incremental-ingestion-backend/src/database/IncrementalIngestionDatabaseManager.ts @@ -28,6 +28,7 @@ import { MarkRecordInsert, } from '../types'; +/** @public */ export class IncrementalIngestionDatabaseManager { private client: Knex; @@ -37,7 +38,7 @@ export class IncrementalIngestionDatabaseManager { /** * Performs an update to the ingestion record with matching `id`. - * @param options IngestionRecordUpdate + * @param options - IngestionRecordUpdate */ async updateIngestionRecordById(options: IngestionRecordUpdate) { await this.client.transaction(async tx => { @@ -48,8 +49,8 @@ export class IncrementalIngestionDatabaseManager { /** * Performs an update to the ingestion record with matching provider name. Will only update active records. - * @param provider string - * @param update Partial + * @param provider - string + * @param update - Partial */ async updateIngestionRecordByProvider( provider: string, @@ -65,7 +66,7 @@ export class IncrementalIngestionDatabaseManager { /** * Performs an insert into the `ingestion.ingestions` table with the supplied values. - * @param record IngestionUpsertIFace + * @param record - IngestionUpsertIFace */ async insertIngestionRecord(record: IngestionUpsertIFace) { await this.client.transaction(async tx => { @@ -100,7 +101,7 @@ export class IncrementalIngestionDatabaseManager { /** * Finds the current ingestion record for the named provider. - * @param provider string + * @param provider - string * @returns IngestionRecord | undefined */ async getCurrentIngestionRecord(provider: string) { @@ -116,7 +117,7 @@ export class IncrementalIngestionDatabaseManager { /** * Removes all entries from `ingestion_marks_entities`, `ingestion_marks`, and `ingestions` * for prior ingestions that completed (i.e., have a `completion_ticket` value other than 'open'). - * @param provider string + * @param provider - string * @returns A count of deletions for each record type. */ async clearFinishedIngestions(provider: string) { @@ -165,8 +166,8 @@ export class IncrementalIngestionDatabaseManager { * 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 + * @param ingestionId - string + * @param provider - string */ async clearDuplicateIngestions(ingestionId: string, provider: string) { await this.client.transaction(async tx => { @@ -195,7 +196,7 @@ export class IncrementalIngestionDatabaseManager { /** * This method fully purges and resets all ingestion records for the named provider, and * leaves it in a paused state. - * @param provider string + * @param provider - string * @returns Counts of all deleted ingestion records */ async purgeAndResetProvider(provider: string) { @@ -262,7 +263,7 @@ export class IncrementalIngestionDatabaseManager { /** * Creates a new ingestion record. - * @param provider string + * @param provider - string * @returns A new ingestion record */ async createProviderIngestionRecord(provider: string) { @@ -285,8 +286,8 @@ export class IncrementalIngestionDatabaseManager { /** * Computes which entities to remove, if any, at the end of a burst. - * @param provider string - * @param ingestionId string + * @param provider - string + * @param ingestionId - string * @returns All entities to remove for this burst. */ async computeRemoved(provider: string, ingestionId: string) { @@ -341,7 +342,7 @@ export class IncrementalIngestionDatabaseManager { /** * Skips any wait time for the next action to run. - * @param provider string + * @param provider - string */ async triggerNextProviderAction(provider: string) { await this.updateIngestionRecordByProvider(provider, { @@ -390,7 +391,7 @@ export class IncrementalIngestionDatabaseManager { /** * Configures the current ingestion record to ingest a burst. - * @param ingestionId string + * @param ingestionId - string */ async setProviderIngesting(ingestionId: string) { await this.updateIngestionRecordById({ @@ -401,7 +402,7 @@ export class IncrementalIngestionDatabaseManager { /** * Indicates the provider is currently ingesting a burst. - * @param ingestionId string + * @param ingestionId - string */ async setProviderBursting(ingestionId: string) { await this.updateIngestionRecordById({ @@ -412,7 +413,7 @@ export class IncrementalIngestionDatabaseManager { /** * Finalizes the current ingestion record to indicate that the post-ingestion rest period is complete. - * @param ingestionId string + * @param ingestionId - string */ async setProviderComplete(ingestionId: string) { await this.updateIngestionRecordById({ @@ -428,8 +429,8 @@ export class IncrementalIngestionDatabaseManager { /** * Marks ingestion as complete and starts the post-ingestion rest cycle. - * @param ingestionId string - * @param restLength Duration + * @param ingestionId - string + * @param restLength - Duration */ async setProviderResting(ingestionId: string, restLength: Duration) { await this.updateIngestionRecordById({ @@ -445,7 +446,7 @@ export class IncrementalIngestionDatabaseManager { /** * Marks ingestion as paused after a burst completes. - * @param ingestionId string + * @param ingestionId - string */ async setProviderInterstitial(ingestionId: string) { await this.updateIngestionRecordById({ @@ -456,8 +457,8 @@ export class IncrementalIngestionDatabaseManager { /** * Starts the cancel process for the current ingestion. - * @param ingestionId string - * @param message string (optional) + * @param ingestionId - string + * @param message - string (optional) */ async setProviderCanceling(ingestionId: string, message?: string) { const update: Partial = { @@ -471,7 +472,7 @@ export class IncrementalIngestionDatabaseManager { /** * Completes the cancel process and triggers a new ingestion. - * @param ingestionId string + * @param ingestionId - string */ async setProviderCanceled(ingestionId: string) { await this.updateIngestionRecordById({ @@ -487,10 +488,10 @@ export class IncrementalIngestionDatabaseManager { /** * 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 + * @param ingestionId - string + * @param attempts - number + * @param error - Error + * @param backoffLength - number */ async setProviderBackoff( ingestionId: string, @@ -512,7 +513,7 @@ export class IncrementalIngestionDatabaseManager { /** * Returns the last record from `ingestion.ingestion_marks` for the supplied ingestionId. - * @param ingestionId string + * @param ingestionId - string * @returns MarkRecord | undefined */ async getLastMark(ingestionId: string) { @@ -536,7 +537,7 @@ export class IncrementalIngestionDatabaseManager { /** * Performs an insert into the `ingestion.ingestion_marks` table with the supplied values. - * @param options MarkRecordInsert + * @param options - MarkRecordInsert */ async createMark(options: MarkRecordInsert) { const { record } = options; @@ -546,8 +547,8 @@ export class IncrementalIngestionDatabaseManager { } /** * Performs an upsert to the `ingestion.ingestion_mark_entities` table for all deferred entities. - * @param markId string - * @param entities DeferredEntity[] + * @param markId - string + * @param entities - DeferredEntity[] */ async createMarkEntities(markId: string, entities: DeferredEntity[]) { await this.client.transaction(async tx => { @@ -563,7 +564,7 @@ export class IncrementalIngestionDatabaseManager { /** * Deletes the entire content of a table, and returns the number of records deleted. - * @param table string + * @param table - string * @returns number */ async purgeTable(table: string) { diff --git a/plugins/incremental-ingestion-backend/src/engine/IncrementalIngestionEngine.ts b/plugins/incremental-ingestion-backend/src/engine/IncrementalIngestionEngine.ts index 574d678e90..790e252dbf 100644 --- a/plugins/incremental-ingestion-backend/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/incremental-ingestion-backend/src/engine/IncrementalIngestionEngine.ts @@ -13,6 +13,7 @@ * 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, @@ -26,7 +27,6 @@ import { performance } from 'perf_hooks'; import { Duration, DurationObjectUnits } from 'luxon'; import { v4 } from 'uuid'; -/** @public */ export class IncrementalIngestionEngine implements IterationEngine { restLength: Duration; backoff: DurationObjectUnits[]; @@ -59,97 +59,106 @@ export class IncrementalIngestionEngine implements IterationEngine { async handleNextAction(signal: AbortSignal) { await this.options.ready; - const { ingestionId, nextActionAt, nextAction, attempts } = - await this.getCurrentAction(); + const result = await this.getCurrentAction(); + if (result) { + const { ingestionId, nextActionAt, nextAction, attempts } = result; - switch (nextAction) { - case 'rest': - if (Date.now() > nextActionAt) { - await this.manager.clearFinishedIngestions( - this.options.provider.getProviderName(), - ); - this.options.logger.info( - `incremental-engine: Ingestion ${ingestionId} rest period complete. Ingestion will start again`, - ); - - await this.manager.setProviderComplete(ingestionId); - } else { - this.options.logger.info( - `incremental-engine: Ingestion '${ingestionId}' rest period continuing`, - ); - } - break; - case 'ingest': - try { - await this.manager.setProviderBursting(ingestionId); - const done = await this.ingestOneBurst(ingestionId, signal); - if (done) { - this.options.logger.info( - `incremental-engine: Ingestion '${ingestionId}' complete, transitioning to rest period of ${this.restLength.toHuman()}`, + switch (nextAction) { + case 'rest': + if (Date.now() > nextActionAt) { + await this.manager.clearFinishedIngestions( + this.options.provider.getProviderName(), ); - await this.manager.setProviderResting(ingestionId, this.restLength); - } else { - await this.manager.setProviderInterstitial(ingestionId); this.options.logger.info( - `incremental-engine: Ingestion '${ingestionId}' continuing`, + `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`, ); } - } catch (error) { - if ( - (error as Error).message && - (error as Error).message === 'CANCEL' - ) { + 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}' canceled`, - ); - await this.manager.setProviderCanceling( - ingestionId, - (error as Error).message, + `incremental-engine: Ingestion '${ingestionId}' backoff complete, will attempt to resume`, ); + await this.manager.setProviderIngesting(ingestionId); } 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, + this.options.logger.info( + `incremental-engine: Ingestion '${ingestionId}' backoff continuing`, ); } - } - break; - case 'backoff': - if (Date.now() > nextActionAt) { + break; + case 'cancel': this.options.logger.info( - `incremental-engine: Ingestion '${ingestionId}' backoff complete, will attempt to resume`, + `incremental-engine: Ingestion '${ingestionId}' canceling, will restart`, ); - await this.manager.setProviderIngesting(ingestionId); - } else { - this.options.logger.info( - `incremental-engine: Ingestion '${ingestionId}' backoff continuing`, + await this.manager.setProviderCanceled(ingestionId); + break; + default: + this.options.logger.error( + `incremental-engine: Ingestion '${ingestionId}' received unknown action '${nextAction}'`, ); - } - break; - case 'cancel': - this.options.logger.info( - `incremental-engine: Ingestion '${ingestionId}' canceling, will restart`, - ); - await this.manager.setProviderCanceled(ingestionId); - break; - default: - this.options.logger.error( - `incremental-engine: Ingestion '${ingestionId}' received unknown action '${nextAction}'`, - ); + } + } else { + this.options.logger.error( + `incremental-engine: Engine tried to create duplicate ingestion record for provider '${this.options.provider.getProviderName()}'.`, + ); } } @@ -170,9 +179,11 @@ export class IncrementalIngestionEngine implements IterationEngine { const result = await this.manager.createProviderIngestionRecord( providerName, ); - this.options.logger.info( - `incremental-engine: Ingestion record created: '${result.ingestionId}'`, - ); + if (result) { + this.options.logger.info( + `incremental-engine: Ingestion record created: '${result.ingestionId}'`, + ); + } return result; } diff --git a/plugins/incremental-ingestion-backend/src/types.ts b/plugins/incremental-ingestion-backend/src/types.ts index 5cc0a5072d..7032b13a0c 100644 --- a/plugins/incremental-ingestion-backend/src/types.ts +++ b/plugins/incremental-ingestion-backend/src/types.ts @@ -50,6 +50,7 @@ export const INCREMENTAL_ENTITY_PROVIDER_ANNOTATION = * batches of entities in sequence so that you never need to have more * than a few hundred in memory at a time. * + * @public */ export interface IncrementalEntityProvider { /** @@ -58,12 +59,6 @@ export interface IncrementalEntityProvider { */ getProviderName(): string; - /** - * Return a function to register the end of the metric that was - * initialized before the fisrt burst call. - */ - prometheusRegister?: { markIngestionCompleted: (result: string) => void }; - /** * Return a single page of entities from a specific point in the * ingestion. @@ -89,8 +84,10 @@ export interface IncrementalEntityProvider { } /** - * Value returned by an @{link IncrementalEntityProvider} to provide a + * Value returned by an {@link IncrementalEntityProvider} to provide a * single page of entities to ingest. + * + * @public */ export interface EntityIteratorResult { /** @@ -111,6 +108,7 @@ export interface EntityIteratorResult { entities: DeferredEntity[]; } +/** @public */ export interface IncrementalEntityProviderOptions { /** * Entities are ingested in bursts. This interval determines how @@ -139,6 +137,7 @@ export interface IncrementalEntityProviderOptions { backoff?: DurationObjectUnits[]; } +/** @public */ export type PluginEnvironment = { logger: Logger; database: PluginDatabaseManager; @@ -148,10 +147,12 @@ export type PluginEnvironment = { permissions: PermissionAuthorizer; }; +/** @public */ export interface IterationEngine { taskFn: TaskFunction; } +/** @public */ export interface IterationEngineOptions { logger: Logger; connection: EntityProviderConnection; @@ -164,6 +165,8 @@ export interface IterationEngineOptions { /** * The shape of data inserted into or updated in the `ingestion.ingestions` table. + * + * @public */ export interface IngestionUpsertIFace { /** @@ -222,6 +225,8 @@ export interface IngestionUpsertIFace { /** * This interface is for updating an existing ingestion record. + * + * @public */ export interface IngestionRecordUpdate { ingestionId: string; @@ -230,6 +235,8 @@ export interface IngestionRecordUpdate { /** * The expected response from the `ingestion.ingestion_marks` table. + * + * @public */ export interface MarkRecord { id: string; @@ -241,6 +248,8 @@ export interface MarkRecord { /** * The expected response from the `ingestion.ingestions` table. + * + * @public */ export interface IngestionRecord extends IngestionUpsertIFace { id: string; @@ -253,6 +262,8 @@ export interface IngestionRecord extends IngestionUpsertIFace { /** * This interface supplies all the values for adding an ingestion mark. + * + * @public */ export interface MarkRecordInsert { record: {