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

Fix more divergent code

Browse Source 
Signed-off-by: Damon Kaswell <damon.kaswell1@hp.com>
This commit is contained in:
Damon Kaswell
2022-11-03 08:48:55 -07:00
committed by Fredrik Adelöw
parent cfe3612e08
commit 5b94fdbaf5
3 changed files with 143 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/incremental-ingestion-backend/src/database/IncrementalIngestionDatabaseManager.ts
+32 -31
View File
@@ -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<IngestionUpsertIFace>
* @param provider - string
* @param update - Partial<IngestionUpsertIFace>
*/
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<IngestionUpsertIFace> = {
@@ -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) {
plugins/incremental-ingestion-backend/src/engine/IncrementalIngestionEngine.ts
+93 -82
View File
@@ -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;
}
plugins/incremental-ingestion-backend/src/types.ts
+18 -7
View File
@@ -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<TCursor, TContext> {
/**
@@ -58,12 +59,6 @@ export interface IncrementalEntityProvider<TCursor, TContext> {
*/
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<TCursor, TContext> {
}
/**
* 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<T> {
/**
@@ -111,6 +108,7 @@ export interface EntityIteratorResult<T> {
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: {