From 559a0de98c8290ecdd7f703523b22e69e303e1c0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Fri, 10 Sep 2021 10:39:24 +0200 Subject: [PATCH 01/23] WIP MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- packages/backend/src/plugins/catalog.ts | 22 ++------- .../packages/backend/src/plugins/catalog.ts | 24 ++-------- .../next/DefaultCatalogProcessingEngine.ts | 48 +++++++++++++------ .../src/next/DefaultLocationStore.ts | 21 ++++++++ .../src/next/NextCatalogBuilder.ts | 13 +++++ .../catalog-backend/src/next/NextRouter.ts | 13 ++++- .../database/DefaultProcessingDatabase.ts | 8 ++++ .../src/next/database/types.ts | 16 +++++++ plugins/catalog-backend/src/next/types.ts | 8 ++++ 9 files changed, 117 insertions(+), 56 deletions(-) diff --git a/packages/backend/src/plugins/catalog.ts b/packages/backend/src/plugins/catalog.ts index 055595dcb5..6a903d0df8 100644 --- a/packages/backend/src/plugins/catalog.ts +++ b/packages/backend/src/plugins/catalog.ts @@ -14,10 +14,7 @@ * limitations under the License. */ -import { - CatalogBuilder, - createRouter, -} from '@backstage/plugin-catalog-backend'; +import { CatalogBuilder } from '@backstage/plugin-catalog-backend'; import { Router } from 'express'; import { PluginEnvironment } from '../types'; @@ -25,20 +22,7 @@ export default async function createPlugin( env: PluginEnvironment, ): Promise { const builder = await CatalogBuilder.create(env); - const { - entitiesCatalog, - locationAnalyzer, - processingEngine, - locationService, - } = await builder.build(); - + const { processingEngine, router } = await builder.build(); await processingEngine.start(); - - return await createRouter({ - entitiesCatalog, - locationAnalyzer, - locationService, - logger: env.logger, - config: env.config, - }); + return router; } diff --git a/packages/create-app/templates/default-app/packages/backend/src/plugins/catalog.ts b/packages/create-app/templates/default-app/packages/backend/src/plugins/catalog.ts index 3145e588cc..d1ded511da 100644 --- a/packages/create-app/templates/default-app/packages/backend/src/plugins/catalog.ts +++ b/packages/create-app/templates/default-app/packages/backend/src/plugins/catalog.ts @@ -1,7 +1,4 @@ -import { - CatalogBuilder, - createRouter, -} from '@backstage/plugin-catalog-backend'; +import { CatalogBuilder } from '@backstage/plugin-catalog-backend'; import { Router } from 'express'; import { PluginEnvironment } from '../types'; @@ -9,22 +6,7 @@ export default async function createPlugin( env: PluginEnvironment, ): Promise { const builder = await CatalogBuilder.create(env); - const { - entitiesCatalog, - locationsCatalog, - locationService, - processingEngine, - locationAnalyzer, - } = await builder.build(); - + const { processingEngine, router } = await builder.build(); await processingEngine.start(); - - return await createRouter({ - entitiesCatalog, - locationsCatalog, - locationService, - locationAnalyzer, - logger: env.logger, - config: env.config, - }); + return router; } diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index d33094ec6d..1911f8f90e 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -36,6 +36,7 @@ import { EntityProvider, EntityProviderConnection, EntityProviderMutation, + EntityRefreshOptions, } from './types'; class Connection implements EntityProviderConnection { @@ -43,8 +44,8 @@ class Connection implements EntityProviderConnection { constructor( private readonly config: { - processingDatabase: ProcessingDatabase; id: string; + processingDatabase: ProcessingDatabase; }, ) {} @@ -60,19 +61,24 @@ class Connection implements EntityProviderConnection { items: mutation.entities, }); }); - return; - } - - this.check(mutation.added.map(e => e.entity)); - this.check(mutation.removed.map(e => e.entity)); - await db.transaction(async tx => { - await db.replaceUnprocessedEntities(tx, { - sourceKey: this.config.id, - type: 'delta', - added: mutation.added, - removed: mutation.removed, + } else if (mutation.type === 'delta') { + this.check(mutation.added.map(e => e.entity)); + this.check(mutation.removed.map(e => e.entity)); + await db.transaction(async tx => { + await db.replaceUnprocessedEntities(tx, { + sourceKey: this.config.id, + type: 'delta', + added: mutation.added, + removed: mutation.removed, + }); }); - }); + } else if (mutation.type === 'refresh') { + await db.transaction(async tx => { + await db.refreshUnprocessedEntities(tx, { + match: mutation.match, + }); + }); + } } private check(entities: Entity[]) { @@ -107,8 +113,8 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { for (const provider of this.entityProviders) { await provider.connect( new Connection({ - processingDatabase: this.processingDatabase, id: provider.getProviderName(), + processingDatabase: this.processingDatabase, }), ); } @@ -237,6 +243,20 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { this.stopFunc = undefined; } } + + async refresh(options: EntityRefreshOptions) { + await Promise.all( + this.entityProviders.map(async provider => { + try { + await provider.refresh?.(options); + } catch (e) { + throw new Error( + `Provider ${provider.getProviderName()} failed refresh, ${e}`, + ); + } + }), + ); + } } // Helps wrap the timing and logging behaviors diff --git a/plugins/catalog-backend/src/next/DefaultLocationStore.ts b/plugins/catalog-backend/src/next/DefaultLocationStore.ts index c9ffe9a682..1e77d60418 100644 --- a/plugins/catalog-backend/src/next/DefaultLocationStore.ts +++ b/plugins/catalog-backend/src/next/DefaultLocationStore.ts @@ -19,10 +19,12 @@ import { ConflictError, NotFoundError } from '@backstage/errors'; import { Knex } from 'knex'; import { v4 as uuid } from 'uuid'; import { DbLocationsRow } from './database/tables'; +import { RefreshStateMatch } from './database/types'; import { getEntityLocationRef } from './processing/util'; import { EntityProvider, EntityProviderConnection, + EntityRefreshOptions, LocationStore, } from './types'; import { locationSpecToLocationEntity } from './util'; @@ -135,6 +137,25 @@ export class DefaultLocationStore implements LocationStore, EntityProvider { }); } + async refresh(options: EntityRefreshOptions) { + const match: RefreshStateMatch = {}; + + // locationKey?: string; + // entityRef?: string; + + if (options.entityRef) { + match.entityRef = options.entityRef; + } + if (options.locationRef) { + // TODO + } + + await this.connection.applyMutation({ + type: 'refresh', + match, + }); + } + private async locations(dbOrTx: Knex.Transaction | Knex = this.db) { const locations = await dbOrTx('locations').select(); return ( diff --git a/plugins/catalog-backend/src/next/NextCatalogBuilder.ts b/plugins/catalog-backend/src/next/NextCatalogBuilder.ts index fa43286777..9186fc3efe 100644 --- a/plugins/catalog-backend/src/next/NextCatalogBuilder.ts +++ b/plugins/catalog-backend/src/next/NextCatalogBuilder.ts @@ -27,6 +27,7 @@ import { } from '@backstage/catalog-model'; import { ScmIntegrations } from '@backstage/integration'; import { createHash } from 'crypto'; +import { Router } from 'express'; import lodash from 'lodash'; import { DatabaseLocationsCatalog, @@ -75,6 +76,7 @@ import { RefreshIntervalFunction, } from './refresh'; import { CatalogEnvironment } from '../service/CatalogBuilder'; +import { createNextRouter } from './NextRouter'; /** * A builder that helps wire up all of the component parts of the catalog. @@ -277,6 +279,7 @@ export class NextCatalogBuilder { locationAnalyzer: LocationAnalyzer; processingEngine: CatalogProcessingEngine; locationService: LocationService; + router: Router; }> { const { config, database, logger } = this.env; @@ -333,12 +336,22 @@ export class NextCatalogBuilder { orchestrator, ); + const router = await createNextRouter({ + entitiesCatalog, + locationAnalyzer, + locationService, + processingEngine, + logger, + config, + }); + return { entitiesCatalog, locationsCatalog, locationAnalyzer, processingEngine, locationService, + router, }; } diff --git a/plugins/catalog-backend/src/next/NextRouter.ts b/plugins/catalog-backend/src/next/NextRouter.ts index 99073be421..9e95cf6434 100644 --- a/plugins/catalog-backend/src/next/NextRouter.ts +++ b/plugins/catalog-backend/src/next/NextRouter.ts @@ -34,12 +34,13 @@ import { parseEntityTransformParams, } from '../service/request'; import { disallowReadonlyMode, validateRequestBody } from '../service/util'; -import { LocationService } from './types'; +import { CatalogProcessingEngine, LocationService, EntityRefreshOptions } from './types'; export interface NextRouterOptions { entitiesCatalog?: EntitiesCatalog; locationAnalyzer?: LocationAnalyzer; locationService: LocationService; + processingEngine?: CatalogProcessingEngine; logger: Logger; config: Config; } @@ -47,7 +48,7 @@ export interface NextRouterOptions { export async function createNextRouter( options: NextRouterOptions, ): Promise { - const { entitiesCatalog, locationAnalyzer, locationService, config, logger } = + const { entitiesCatalog, locationAnalyzer, locationService, processingEngine, config, logger } = options; const router = Router(); @@ -59,6 +60,14 @@ export async function createNextRouter( logger.info('Catalog is running in readonly mode'); } + if (processingEngine) { + router.post('/refresh'), async (req, res) => { + const options: EntityRefreshOptions = req.body; + await processingEngine.refresh(options); + res.status(200); + }); + } + if (entitiesCatalog) { router .get('/entities', async (req, res) => { diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index 30457f2cbd..d2c17f7ad1 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -36,6 +36,7 @@ import { GetProcessableEntitiesResult, ProcessingDatabase, RefreshStateItem, + RefreshUnprocessedEntitiesOptions, ReplaceUnprocessedEntitiesOptions, UpdateProcessedEntityOptions, } from './types'; @@ -508,6 +509,13 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { }; } + async refreshUnprocessedEntities( + txOpaque: Transaction, + options: RefreshUnprocessedEntitiesOptions, + ): Promise { + const tx = txOpaque as Knex.Transaction; + } + async transaction(fn: (tx: Transaction) => Promise): Promise { try { let result: T | undefined = undefined; diff --git a/plugins/catalog-backend/src/next/database/types.ts b/plugins/catalog-backend/src/next/database/types.ts index 38ca59a431..3a65cc7361 100644 --- a/plugins/catalog-backend/src/next/database/types.ts +++ b/plugins/catalog-backend/src/next/database/types.ts @@ -79,6 +79,16 @@ export type ReplaceUnprocessedEntitiesOptions = type: 'delta'; }; +export type RefreshStateMatch = { + locationKey?: string; + entityRef?: string; + parentOfEntityRef: string; +}; + +export type RefreshUnprocessedEntitiesOptions = { + match: RefreshStateMatch; +}; + export interface ProcessingDatabase { transaction(fn: (tx: Transaction) => Promise): Promise; @@ -86,6 +96,7 @@ export interface ProcessingDatabase { txOpaque: Transaction, options: ReplaceUnprocessedEntitiesOptions, ): Promise; + getProcessableEntities( txOpaque: Transaction, request: { processBatchSize: number }, @@ -106,4 +117,9 @@ export interface ProcessingDatabase { txOpaque: Transaction, options: UpdateProcessedEntityErrorsOptions, ): Promise; + + refreshUnprocessedEntities( + txOpaque: Transaction, + options: RefreshUnprocessedEntitiesOptions, + ): Promise; } diff --git a/plugins/catalog-backend/src/next/types.ts b/plugins/catalog-backend/src/next/types.ts index 3b27dd7d10..0c34a873ff 100644 --- a/plugins/catalog-backend/src/next/types.ts +++ b/plugins/catalog-backend/src/next/types.ts @@ -15,6 +15,7 @@ */ import { Entity, Location, LocationSpec } from '@backstage/catalog-model'; +import { RefreshStateMatch } from './database/types'; import { DeferredEntity } from './processing/types'; export interface LocationService { @@ -37,9 +38,15 @@ export interface LocationStore { export interface CatalogProcessingEngine { start(): Promise; stop(): Promise; + refresh(options: EntityRefreshOptions): Promise; } +export type EntityRefreshOptions = + | { entityRef: string } // example: component:default/backstage + | { locationRef: string }; // example: url:https://github.com/backstage/backstage/blob/master/catalog-info.yaml + export type EntityProviderMutation = + | { type: 'refresh'; match: RefreshStateMatch } // TODO(jhaals): Should this really use a type from the db? | { type: 'full'; entities: DeferredEntity[] } | { type: 'delta'; added: DeferredEntity[]; removed: DeferredEntity[] }; @@ -50,4 +57,5 @@ export interface EntityProviderConnection { export interface EntityProvider { getProviderName(): string; connect(connection: EntityProviderConnection): Promise; + refresh?(options: EntityRefreshOptions): Promise; } From 20cce38c844e0227940e13a5e9c70b8e385749ba Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Fri, 10 Sep 2021 14:19:19 +0200 Subject: [PATCH 02/23] Fix refresh route Co-authored-by: Ben Lambert Signed-off-by: Johan Haals --- .../catalog-backend/src/next/NextRouter.ts | 24 +++++++++++++------ 1 file changed, 17 insertions(+), 7 deletions(-) diff --git a/plugins/catalog-backend/src/next/NextRouter.ts b/plugins/catalog-backend/src/next/NextRouter.ts index 9e95cf6434..03d4005769 100644 --- a/plugins/catalog-backend/src/next/NextRouter.ts +++ b/plugins/catalog-backend/src/next/NextRouter.ts @@ -34,7 +34,11 @@ import { parseEntityTransformParams, } from '../service/request'; import { disallowReadonlyMode, validateRequestBody } from '../service/util'; -import { CatalogProcessingEngine, LocationService, EntityRefreshOptions } from './types'; +import { + CatalogProcessingEngine, + LocationService, + EntityRefreshOptions, +} from './types'; export interface NextRouterOptions { entitiesCatalog?: EntitiesCatalog; @@ -48,8 +52,14 @@ export interface NextRouterOptions { export async function createNextRouter( options: NextRouterOptions, ): Promise { - const { entitiesCatalog, locationAnalyzer, locationService, processingEngine, config, logger } = - options; + const { + entitiesCatalog, + locationAnalyzer, + locationService, + processingEngine, + config, + logger, + } = options; const router = Router(); router.use(express.json()); @@ -61,10 +71,10 @@ export async function createNextRouter( } if (processingEngine) { - router.post('/refresh'), async (req, res) => { - const options: EntityRefreshOptions = req.body; - await processingEngine.refresh(options); - res.status(200); + router.post('/refresh', async (req, res) => { + const refreshOptions: EntityRefreshOptions = req.body; + await processingEngine.refresh(refreshOptions); + res.status(200).send(); }); } From 9c9353e2d3e22bb4aee0db82a2beed447f1158b5 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Fri, 10 Sep 2021 14:20:15 +0200 Subject: [PATCH 03/23] WIP refreshing entities Signed-off-by: Johan Haals --- .../next/DefaultCatalogProcessingEngine.ts | 36 ++++---- .../src/next/DefaultLocationStore.ts | 31 ++++--- .../DefaultProcessingDatabase.test.ts | 83 +++++++++++++++++++ .../database/DefaultProcessingDatabase.ts | 60 +++++++++++++- .../src/next/database/types.ts | 9 +- 5 files changed, 181 insertions(+), 38 deletions(-) diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index 1911f8f90e..0e77e6e9b4 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -73,11 +73,12 @@ class Connection implements EntityProviderConnection { }); }); } else if (mutation.type === 'refresh') { - await db.transaction(async tx => { - await db.refreshUnprocessedEntities(tx, { - match: mutation.match, - }); - }); + // await db.transaction(async tx => { + // await db.refreshUnprocessedEntities(tx, { + // match: mutation.match, + // }); + // }); + console.log('wopoop'); } } @@ -245,17 +246,20 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { } async refresh(options: EntityRefreshOptions) { - await Promise.all( - this.entityProviders.map(async provider => { - try { - await provider.refresh?.(options); - } catch (e) { - throw new Error( - `Provider ${provider.getProviderName()} failed refresh, ${e}`, - ); - } - }), - ); + await this.processingDatabase.transaction(async tx => { + await this.processingDatabase.refreshUnprocessedEntities(tx, options); + }); + // await Promise.all( + // this.entityProviders.map(async provider => { + // try { + // await provider.refresh?.(options); + // } catch (e) { + // throw new Error( + // `Provider ${provider.getProviderName()} failed refresh, ${e}`, + // ); + // } + // }), + // ); } } diff --git a/plugins/catalog-backend/src/next/DefaultLocationStore.ts b/plugins/catalog-backend/src/next/DefaultLocationStore.ts index 1e77d60418..09eacc3955 100644 --- a/plugins/catalog-backend/src/next/DefaultLocationStore.ts +++ b/plugins/catalog-backend/src/next/DefaultLocationStore.ts @@ -137,23 +137,20 @@ export class DefaultLocationStore implements LocationStore, EntityProvider { }); } - async refresh(options: EntityRefreshOptions) { - const match: RefreshStateMatch = {}; - - // locationKey?: string; - // entityRef?: string; - - if (options.entityRef) { - match.entityRef = options.entityRef; - } - if (options.locationRef) { - // TODO - } - - await this.connection.applyMutation({ - type: 'refresh', - match, - }); + async refresh(_options: EntityRefreshOptions) { + // const match: RefreshStateMatch = {}; + // // locationKey?: string; + // // entityRef?: string; + // if (options.entityRef) { + // match.entityRef = options.entityRef; + // } + // if (options.locationRef) { + // // TODO + // } + // await this.connection.applyMutation({ + // type: 'refresh', + // match, + // }); } private async locations(dbOrTx: Knex.Transaction | Knex = this.db) { diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts index 23829bd33f..e027830601 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts @@ -1017,4 +1017,87 @@ describe('Default Processing Database', () => { 60_000, ); }); + + describe('refreshEntities', () => { + it.each(databases.eachSupportedId())( + 'should refresh the location further up the tree, %p', + async databaseId => { + const { knex, db } = await createDatabase(databaseId); + + await knex('refresh_state').insert({ + entity_id: '7', + entity_ref: 'location:default/myloc', + unprocessed_entity: JSON.stringify({ + kind: 'Location', + apiVersion: '1.0.0', + metadata: { + name: 'xyz', + }, + } as Entity), + errors: '[]', + next_update_at: '2031-01-01 23:00:00', + last_discovery_at: '2021-04-01 13:37:00', + }); + + await knex('refresh_state').insert({ + entity_id: '8', + entity_ref: 'component:default/mycomp', + unprocessed_entity: JSON.stringify({ + kind: 'Component', + apiVersion: '1.0.0', + metadata: { + name: 'xyz', + }, + } as Entity), + errors: '[]', + next_update_at: '2031-01-01 23:00:00', + last_discovery_at: '2021-04-01 13:37:00', + }); + + await knex('refresh_state').insert({ + entity_id: '9', + entity_ref: 'api:default/myapi', + unprocessed_entity: JSON.stringify({ + kind: 'Api', + apiVersion: '1.0.0', + metadata: { + name: 'xyz', + }, + } as Entity), + errors: '[]', + next_update_at: '2031-01-01 23:00:00', + last_discovery_at: '2021-04-01 13:37:00', + }); + + await insertRefRow(knex, { + source_entity_ref: 'component:default/mycomp', + target_entity_ref: 'api:default/myapi', + }); + await insertRefRow(knex, { + source_entity_ref: 'location:default/myloc', + target_entity_ref: 'component:default/mycomp', + }); + + await insertRefRow(knex, { + source_key: 'ConfigLocationProvider', + target_entity_ref: 'location:default/myloc', + }); + + await db.transaction(async tx => { + await db.refreshUnprocessedEntities(tx, { + entityRef: 'api:default/myapi', + }); + }); + + const [result] = await knex('refresh_state') + .where('entity_ref', 'api:default/myapi') + .select(); + + // TODO: This is going to break after 2031 + expect(parseDate(result.next_update_at).year).toEqual( + DateTime.local().year, + ); + }, + ); + }); }); diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index d2c17f7ad1..a19f88188d 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -14,9 +14,13 @@ * limitations under the License. */ -import { Entity, stringifyEntityRef } from '@backstage/catalog-model'; +import { + Entity, + LOCATION_ANNOTATION, + stringifyEntityRef, +} from '@backstage/catalog-model'; import { JsonObject } from '@backstage/config'; -import { ConflictError } from '@backstage/errors'; +import { ConflictError, NotFoundError } from '@backstage/errors'; import { Knex } from 'knex'; import lodash from 'lodash'; import { v4 as uuid } from 'uuid'; @@ -351,6 +355,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { unprocessed_entity: serializedEntity, location_key: locationKey, last_discovery_at: tx.fn.now(), + next_update_at: tx.fn.now(), }) .where('entity_ref', entityRef) .andWhere(inner => { @@ -509,11 +514,62 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { }; } + // TODO(jhaals): Rename this to refreshEntity? async refreshUnprocessedEntities( txOpaque: Transaction, options: RefreshUnprocessedEntitiesOptions, ): Promise { const tx = txOpaque as Knex.Transaction; + if ('entityRef' in options) { + const { entityRef } = options; + const [result] = await tx('refresh_state') + .where({ entity_ref: entityRef }) + .select(); + + if (!result) { + throw new NotFoundError(`EntityRef ${entityRef} not found`); + } + + const refs = await tx( + 'refresh_state_references', + ) + .where({ target_entity_ref: entityRef }) + .select(); + + for (const ref of refs) { + if (ref.source_entity_ref?.startsWith('location:')) { + const updateResult = await tx('refresh_state') + .where({ entity_ref: ref.source_entity_ref }) + .update({ next_update_at: tx.fn.now() }); + + if (updateResult === 0) { + throw new ConflictError( + `Failed to schedule ${ref.source_entity_ref} for refresh`, + ); + } + } + } + /* + For a given entityRef: + - Fetch entity + - recursively select from refresh_state_references where target is our entityRef. + Continue until we find a location. + - Process and run addUnprocessedEntities with a flag telling it to bump the timestamp for all deferred entities. + */ + /* + For a given URL + - Fetch entity based on managed by location URL? + - repeat for entityRef + */ + } + + if ('locationRef' in options) { + // TODO(jhaals): managed by or origin? + // const entity: Entity = JSON.parse(result.processed_entity); + // // TODO: ONly required for URLz + // const managedBy = entity.metadata?.annotations?.[LOCATION_ANNOTATION] + // console.log(managedBy); + } } async transaction(fn: (tx: Transaction) => Promise): Promise { diff --git a/plugins/catalog-backend/src/next/database/types.ts b/plugins/catalog-backend/src/next/database/types.ts index 3a65cc7361..e8ced9d250 100644 --- a/plugins/catalog-backend/src/next/database/types.ts +++ b/plugins/catalog-backend/src/next/database/types.ts @@ -85,9 +85,12 @@ export type RefreshStateMatch = { parentOfEntityRef: string; }; -export type RefreshUnprocessedEntitiesOptions = { - match: RefreshStateMatch; -}; +export type RefreshUnprocessedEntitiesOptions = + | { + // match: RefreshStateMatch; + entityRef: string; + } + | { locationRef: string }; export interface ProcessingDatabase { transaction(fn: (tx: Transaction) => Promise): Promise; From b7497a992c6c9453b50c61ff923bc6ddfd918209 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Mon, 13 Sep 2021 11:34:13 +0200 Subject: [PATCH 04/23] catalog-backend: tests for processingEngine refresh Co-authored-by: Patrik Oldsberg Signed-off-by: Johan Haals --- .../DefaultCatalogProcessingEngine.test.ts | 242 +++++++++++++++++- .../next/DefaultCatalogProcessingEngine.ts | 2 + .../DefaultProcessingDatabase.test.ts | 83 ------ .../database/DefaultProcessingDatabase.ts | 2 +- 4 files changed, 243 insertions(+), 86 deletions(-) diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts index f7dede4adb..a0d62c8642 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts @@ -15,13 +15,27 @@ */ import { getVoidLogger } from '@backstage/backend-common'; -import { Hash } from 'crypto'; +import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils'; +import { createHash, Hash } from 'crypto'; +import { Knex } from 'knex'; +import { Logger } from 'winston'; import { DateTime } from 'luxon'; +import { DatabaseManager } from './database/DatabaseManager'; import waitForExpect from 'wait-for-expect'; import { DefaultProcessingDatabase } from './database/DefaultProcessingDatabase'; +import { + DbRefreshStateReferencesRow, + DbRefreshStateRow, +} from './database/tables'; +import { ProcessingDatabase } from './database/types'; import { DefaultCatalogProcessingEngine } from './DefaultCatalogProcessingEngine'; -import { CatalogProcessingOrchestrator } from './processing/types'; +import { + CatalogProcessingOrchestrator, + EntityProcessingRequest, +} from './processing/types'; import { Stitcher } from './stitching/Stitcher'; +import { Entity, stringifyEntityRef } from '@backstage/catalog-model'; +import { v4 as uuid } from 'uuid'; describe('DefaultCatalogProcessingEngine', () => { const db = { @@ -236,3 +250,227 @@ describe('DefaultCatalogProcessingEngine', () => { await engine.stop(); }); }); + +describe('DefaultCatalogProcessingEngine integration', () => { + const defaultLogger = getVoidLogger(); + const databases = TestDatabases.create({ + ids: ['POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'], + }); + + async function createDatabase( + databaseId: TestDatabaseId, + logger: Logger = defaultLogger, + ) { + const knex = await databases.init(databaseId); + await DatabaseManager.createDatabase(knex); + return { + knex, + db: new DefaultProcessingDatabase({ + database: knex, + logger, + refreshInterval: () => 100, + }), + }; + } + + const createPopulatedEngine = async (options: { + db: ProcessingDatabase; + knex: Knex; + entities: Entity[]; + references: { [source: string]: string[] }; + }) => { + const { db, knex, entities, references } = options; + + const entityMap = new Map( + entities.map(entity => [stringifyEntityRef(entity), entity]), + ); + + for (const entity of entities) { + await knex('refresh_state').insert({ + entity_id: uuid(), + entity_ref: stringifyEntityRef(entity), + unprocessed_entity: JSON.stringify(entity), + errors: '[]', + next_update_at: '2031-01-01 23:00:00', + last_discovery_at: '2021-04-01 13:37:00', + }); + } + + for (const entityRef of entityMap.keys()) { + if (!(entityRef in references)) { + await knex( + 'refresh_state_references', + ).insert({ + source_key: 'ConfigLocationProvider', + target_entity_ref: entityRef, + }); + } + } + for (const [sourceRef, targetRefs] of Object.entries(references)) { + for (const targetRef of targetRefs) { + await knex( + 'refresh_state_references', + ).insert({ + source_entity_ref: sourceRef, + target_entity_ref: targetRef, + }); + } + } + + const engine = new DefaultCatalogProcessingEngine( + defaultLogger, + [], + db, + { + async process(request: EntityProcessingRequest) { + const entityRef = stringifyEntityRef(request.entity); + const entity = entityMap.get(entityRef); + if (!entity) { + throw new Error(`Unexpected entity: ${entityRef}`); + } + const deferredEntities = + references[entityRef]?.map(ref => { + const e = entityMap.get(ref); + if (!e) { + throw new Error(`Target entity not found: ${ref}`); + } + return { entity: e, locationKey: ref }; + }) || []; + + return { + ok: true, + completedEntity: { + ...entity, + metadata: { + ...entity.metadata, + annotations: { + ...entity.metadata.annotations, + 'refresh-completed': 'true', + }, + }, + }, + relations: [], + errors: [], + deferredEntities, + state: new Map(), + }; + }, + }, + new Stitcher(knex, defaultLogger), + () => createHash('sha1'), + 50, + ); + + return engine; + }; + + const waitForRefresh = async (knex: Knex, entityRef: string) => { + for (;;) { + const [result] = await knex('refresh_state') + .where('entity_ref', entityRef) + .select(); + + const entity = result.processed_entity + ? (JSON.parse(result.processed_entity) as Entity) + : undefined; + if (entity?.metadata?.annotations?.['refresh-completed']) { + return true; + } + await new Promise(resolve => setTimeout(resolve, 500)); + } + }; + + it.each(databases.eachSupportedId())( + 'should refresh the parent location, %p', + async databaseId => { + const { knex, db } = await createDatabase(databaseId); + + const engine = await createPopulatedEngine({ + db, + knex, + entities: [ + { + kind: 'Location', + apiVersion: '1.0.0', + metadata: { + name: 'myloc', + }, + }, + { + kind: 'Component', + apiVersion: '1.0.0', + metadata: { + name: 'mycomp', + }, + }, + ], + references: { + 'location:default/myloc': ['component:default/mycomp'], + }, + }); + + await engine.start(); + + await engine.refresh({ + entityRef: 'component:default/mycomp', + }); + + await expect( + waitForRefresh(knex, 'component:default/mycomp'), + ).resolves.toBe(true); + + await engine.stop(); + }, + ); + + it.each(databases.eachSupportedId())( + 'should refresh the location further up the tree, %p', + async databaseId => { + const { knex, db } = await createDatabase(databaseId); + + const engine = await createPopulatedEngine({ + db, + knex, + entities: [ + { + kind: 'Location', + apiVersion: '1.0.0', + metadata: { + name: 'myloc', + }, + }, + { + kind: 'Component', + apiVersion: '1.0.0', + metadata: { + name: 'mycomp', + }, + }, + { + kind: 'Api', + apiVersion: '1.0.0', + metadata: { + name: 'myapi', + }, + }, + ], + references: { + 'location:default/myloc': ['component:default/mycomp'], + 'component:default/mycomp': ['api:default/myapi'], + }, + }); + + await engine.start(); + + await engine.refresh({ + entityRef: 'api:default/myapi', + }); + + await expect(waitForRefresh(knex, 'api:default/myapi')).resolves.toBe( + true, + ); + + await engine.stop(); + }, + ); +}); diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index 0e77e6e9b4..b1cfc33885 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -104,6 +104,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { private readonly orchestrator: CatalogProcessingOrchestrator, private readonly stitcher: Stitcher, private readonly createHash: () => Hash, + private readonly pollingIntervalMs: number = 1000, ) {} async start() { @@ -123,6 +124,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { this.stopFunc = startTaskPipeline({ lowWatermark: 5, highWatermark: 10, + pollingIntervalMs: this.pollingIntervalMs, loadTasks: async count => { try { const { items } = await this.processingDatabase.transaction( diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts index e027830601..23829bd33f 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts @@ -1017,87 +1017,4 @@ describe('Default Processing Database', () => { 60_000, ); }); - - describe('refreshEntities', () => { - it.each(databases.eachSupportedId())( - 'should refresh the location further up the tree, %p', - async databaseId => { - const { knex, db } = await createDatabase(databaseId); - - await knex('refresh_state').insert({ - entity_id: '7', - entity_ref: 'location:default/myloc', - unprocessed_entity: JSON.stringify({ - kind: 'Location', - apiVersion: '1.0.0', - metadata: { - name: 'xyz', - }, - } as Entity), - errors: '[]', - next_update_at: '2031-01-01 23:00:00', - last_discovery_at: '2021-04-01 13:37:00', - }); - - await knex('refresh_state').insert({ - entity_id: '8', - entity_ref: 'component:default/mycomp', - unprocessed_entity: JSON.stringify({ - kind: 'Component', - apiVersion: '1.0.0', - metadata: { - name: 'xyz', - }, - } as Entity), - errors: '[]', - next_update_at: '2031-01-01 23:00:00', - last_discovery_at: '2021-04-01 13:37:00', - }); - - await knex('refresh_state').insert({ - entity_id: '9', - entity_ref: 'api:default/myapi', - unprocessed_entity: JSON.stringify({ - kind: 'Api', - apiVersion: '1.0.0', - metadata: { - name: 'xyz', - }, - } as Entity), - errors: '[]', - next_update_at: '2031-01-01 23:00:00', - last_discovery_at: '2021-04-01 13:37:00', - }); - - await insertRefRow(knex, { - source_entity_ref: 'component:default/mycomp', - target_entity_ref: 'api:default/myapi', - }); - await insertRefRow(knex, { - source_entity_ref: 'location:default/myloc', - target_entity_ref: 'component:default/mycomp', - }); - - await insertRefRow(knex, { - source_key: 'ConfigLocationProvider', - target_entity_ref: 'location:default/myloc', - }); - - await db.transaction(async tx => { - await db.refreshUnprocessedEntities(tx, { - entityRef: 'api:default/myapi', - }); - }); - - const [result] = await knex('refresh_state') - .where('entity_ref', 'api:default/myapi') - .select(); - - // TODO: This is going to break after 2031 - expect(parseDate(result.next_update_at).year).toEqual( - DateTime.local().year, - ); - }, - ); - }); }); diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index a19f88188d..369deb36a4 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -514,7 +514,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { }; } - // TODO(jhaals): Rename this to refreshEntity? + // TODO(jhaals): Rename this to refreshEntity/refreshEntities? async refreshUnprocessedEntities( txOpaque: Transaction, options: RefreshUnprocessedEntitiesOptions, From 40a40c2fb2d02d0ead326d48f2811a5471fec58c Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Mon, 13 Sep 2021 13:08:55 +0200 Subject: [PATCH 05/23] Implement refresh for nested entities Co-authored-by: Patrik Oldsberg Signed-off-by: Johan Haals --- .../database/DefaultProcessingDatabase.ts | 69 ++++++++++++------- 1 file changed, 46 insertions(+), 23 deletions(-) diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index 369deb36a4..58a384457d 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -50,6 +50,7 @@ import { // errors in the underlying engine due to exceeding query limits, but large // enough to get the speed benefits. const BATCH_SIZE = 50; +const MAX_REFRESH_DEPTH = 10; export class DefaultProcessingDatabase implements ProcessingDatabase { constructor( @@ -522,33 +523,55 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { const tx = txOpaque as Knex.Transaction; if ('entityRef' in options) { const { entityRef } = options; - const [result] = await tx('refresh_state') - .where({ entity_ref: entityRef }) - .select(); - if (!result) { - throw new NotFoundError(`EntityRef ${entityRef} not found`); - } + let refreshTarget = entityRef; - const refs = await tx( - 'refresh_state_references', - ) - .where({ target_entity_ref: entityRef }) - .select(); - - for (const ref of refs) { - if (ref.source_entity_ref?.startsWith('location:')) { - const updateResult = await tx('refresh_state') - .where({ entity_ref: ref.source_entity_ref }) - .update({ next_update_at: tx.fn.now() }); - - if (updateResult === 0) { - throw new ConflictError( - `Failed to schedule ${ref.source_entity_ref} for refresh`, - ); - } + let currentRef = entityRef; + let depth = 0; + for (;;) { + if (depth++ > MAX_REFRESH_DEPTH) { + throw new Error( + `Unable to refresh Entity ${entityRef}, maximum refresh depth of ${MAX_REFRESH_DEPTH} reached`, + ); } + + const rows = await tx( + 'refresh_state_references', + ) + .where({ target_entity_ref: currentRef }) + .select(); + + if (rows.length === 0) { + if (depth === 1) { + throw new NotFoundError(`Entity ${currentRef} not found`); + } + throw new NotFoundError( + `Entity ${entityRef} has a broken parent reference chain at ${currentRef}`, + ); + } + + const parentRef = rows[0].source_entity_ref; + if (!parentRef) { + // We've reached the top of the tree which is the entityProvider. + // In this case we refresh the entity itself. + break; + } + if (parentRef.startsWith('location:')) { + refreshTarget = parentRef; + break; + } + currentRef = parentRef; } + + const updateResult = await tx('refresh_state') + .where({ entity_ref: refreshTarget }) + .update({ next_update_at: tx.fn.now() }); + if (updateResult === 0) { + throw new ConflictError( + `Failed to schedule ${refreshTarget} for refresh`, + ); + } + /* For a given entityRef: - Fetch entity From b7b66448c039f8583d4deb82b49b864165cd6bc4 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Mon, 13 Sep 2021 14:22:36 +0200 Subject: [PATCH 06/23] Rename refresh methods, cleanup previous code Co-authored-by: Patrik Oldsberg Signed-off-by: Johan Haals --- .../next/DefaultCatalogProcessingEngine.ts | 20 +------------------ .../src/next/DefaultLocationStore.ts | 18 ----------------- .../database/DefaultProcessingDatabase.ts | 14 +++---------- .../src/next/database/types.ts | 18 +++-------------- plugins/catalog-backend/src/next/types.ts | 7 +------ 5 files changed, 8 insertions(+), 69 deletions(-) diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index b1cfc33885..6e3f6589d2 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -72,13 +72,6 @@ class Connection implements EntityProviderConnection { removed: mutation.removed, }); }); - } else if (mutation.type === 'refresh') { - // await db.transaction(async tx => { - // await db.refreshUnprocessedEntities(tx, { - // match: mutation.match, - // }); - // }); - console.log('wopoop'); } } @@ -249,19 +242,8 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { async refresh(options: EntityRefreshOptions) { await this.processingDatabase.transaction(async tx => { - await this.processingDatabase.refreshUnprocessedEntities(tx, options); + await this.processingDatabase.refresh(tx, options); }); - // await Promise.all( - // this.entityProviders.map(async provider => { - // try { - // await provider.refresh?.(options); - // } catch (e) { - // throw new Error( - // `Provider ${provider.getProviderName()} failed refresh, ${e}`, - // ); - // } - // }), - // ); } } diff --git a/plugins/catalog-backend/src/next/DefaultLocationStore.ts b/plugins/catalog-backend/src/next/DefaultLocationStore.ts index 09eacc3955..c9ffe9a682 100644 --- a/plugins/catalog-backend/src/next/DefaultLocationStore.ts +++ b/plugins/catalog-backend/src/next/DefaultLocationStore.ts @@ -19,12 +19,10 @@ import { ConflictError, NotFoundError } from '@backstage/errors'; import { Knex } from 'knex'; import { v4 as uuid } from 'uuid'; import { DbLocationsRow } from './database/tables'; -import { RefreshStateMatch } from './database/types'; import { getEntityLocationRef } from './processing/util'; import { EntityProvider, EntityProviderConnection, - EntityRefreshOptions, LocationStore, } from './types'; import { locationSpecToLocationEntity } from './util'; @@ -137,22 +135,6 @@ export class DefaultLocationStore implements LocationStore, EntityProvider { }); } - async refresh(_options: EntityRefreshOptions) { - // const match: RefreshStateMatch = {}; - // // locationKey?: string; - // // entityRef?: string; - // if (options.entityRef) { - // match.entityRef = options.entityRef; - // } - // if (options.locationRef) { - // // TODO - // } - // await this.connection.applyMutation({ - // type: 'refresh', - // match, - // }); - } - private async locations(dbOrTx: Knex.Transaction | Knex = this.db) { const locations = await dbOrTx('locations').select(); return ( diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index 58a384457d..e46ad8dbb2 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -14,11 +14,7 @@ * limitations under the License. */ -import { - Entity, - LOCATION_ANNOTATION, - stringifyEntityRef, -} from '@backstage/catalog-model'; +import { Entity, stringifyEntityRef } from '@backstage/catalog-model'; import { JsonObject } from '@backstage/config'; import { ConflictError, NotFoundError } from '@backstage/errors'; import { Knex } from 'knex'; @@ -40,7 +36,7 @@ import { GetProcessableEntitiesResult, ProcessingDatabase, RefreshStateItem, - RefreshUnprocessedEntitiesOptions, + RefreshOptions, ReplaceUnprocessedEntitiesOptions, UpdateProcessedEntityOptions, } from './types'; @@ -515,11 +511,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { }; } - // TODO(jhaals): Rename this to refreshEntity/refreshEntities? - async refreshUnprocessedEntities( - txOpaque: Transaction, - options: RefreshUnprocessedEntitiesOptions, - ): Promise { + async refresh(txOpaque: Transaction, options: RefreshOptions): Promise { const tx = txOpaque as Knex.Transaction; if ('entityRef' in options) { const { entityRef } = options; diff --git a/plugins/catalog-backend/src/next/database/types.ts b/plugins/catalog-backend/src/next/database/types.ts index e8ced9d250..509d9d0d0e 100644 --- a/plugins/catalog-backend/src/next/database/types.ts +++ b/plugins/catalog-backend/src/next/database/types.ts @@ -79,19 +79,10 @@ export type ReplaceUnprocessedEntitiesOptions = type: 'delta'; }; -export type RefreshStateMatch = { - locationKey?: string; - entityRef?: string; - parentOfEntityRef: string; +export type RefreshOptions = { + entityRef: string; }; -export type RefreshUnprocessedEntitiesOptions = - | { - // match: RefreshStateMatch; - entityRef: string; - } - | { locationRef: string }; - export interface ProcessingDatabase { transaction(fn: (tx: Transaction) => Promise): Promise; @@ -121,8 +112,5 @@ export interface ProcessingDatabase { options: UpdateProcessedEntityErrorsOptions, ): Promise; - refreshUnprocessedEntities( - txOpaque: Transaction, - options: RefreshUnprocessedEntitiesOptions, - ): Promise; + refresh(txOpaque: Transaction, options: RefreshOptions): Promise; } diff --git a/plugins/catalog-backend/src/next/types.ts b/plugins/catalog-backend/src/next/types.ts index 0c34a873ff..91c8c47fe4 100644 --- a/plugins/catalog-backend/src/next/types.ts +++ b/plugins/catalog-backend/src/next/types.ts @@ -15,7 +15,6 @@ */ import { Entity, Location, LocationSpec } from '@backstage/catalog-model'; -import { RefreshStateMatch } from './database/types'; import { DeferredEntity } from './processing/types'; export interface LocationService { @@ -41,12 +40,9 @@ export interface CatalogProcessingEngine { refresh(options: EntityRefreshOptions): Promise; } -export type EntityRefreshOptions = - | { entityRef: string } // example: component:default/backstage - | { locationRef: string }; // example: url:https://github.com/backstage/backstage/blob/master/catalog-info.yaml +export type EntityRefreshOptions = { entityRef: string }; export type EntityProviderMutation = - | { type: 'refresh'; match: RefreshStateMatch } // TODO(jhaals): Should this really use a type from the db? | { type: 'full'; entities: DeferredEntity[] } | { type: 'delta'; added: DeferredEntity[]; removed: DeferredEntity[] }; @@ -57,5 +53,4 @@ export interface EntityProviderConnection { export interface EntityProvider { getProviderName(): string; connect(connection: EntityProviderConnection): Promise; - refresh?(options: EntityRefreshOptions): Promise; } From 16e03e3388134b3ed3f714eb9186347303818304 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Mon, 13 Sep 2021 14:28:05 +0200 Subject: [PATCH 07/23] Remove WIP comments Signed-off-by: Johan Haals --- .../database/DefaultProcessingDatabase.ts | 105 +++++++----------- 1 file changed, 41 insertions(+), 64 deletions(-) diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index e46ad8dbb2..3cc1ef8233 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -513,77 +513,54 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { async refresh(txOpaque: Transaction, options: RefreshOptions): Promise { const tx = txOpaque as Knex.Transaction; - if ('entityRef' in options) { - const { entityRef } = options; + const { entityRef } = options; - let refreshTarget = entityRef; + let refreshTarget = entityRef; - let currentRef = entityRef; - let depth = 0; - for (;;) { - if (depth++ > MAX_REFRESH_DEPTH) { - throw new Error( - `Unable to refresh Entity ${entityRef}, maximum refresh depth of ${MAX_REFRESH_DEPTH} reached`, - ); - } - - const rows = await tx( - 'refresh_state_references', - ) - .where({ target_entity_ref: currentRef }) - .select(); - - if (rows.length === 0) { - if (depth === 1) { - throw new NotFoundError(`Entity ${currentRef} not found`); - } - throw new NotFoundError( - `Entity ${entityRef} has a broken parent reference chain at ${currentRef}`, - ); - } - - const parentRef = rows[0].source_entity_ref; - if (!parentRef) { - // We've reached the top of the tree which is the entityProvider. - // In this case we refresh the entity itself. - break; - } - if (parentRef.startsWith('location:')) { - refreshTarget = parentRef; - break; - } - currentRef = parentRef; - } - - const updateResult = await tx('refresh_state') - .where({ entity_ref: refreshTarget }) - .update({ next_update_at: tx.fn.now() }); - if (updateResult === 0) { - throw new ConflictError( - `Failed to schedule ${refreshTarget} for refresh`, + let currentRef = entityRef; + let depth = 0; + for (;;) { + if (depth++ > MAX_REFRESH_DEPTH) { + throw new Error( + `Unable to refresh Entity ${entityRef}, maximum refresh depth of ${MAX_REFRESH_DEPTH} reached`, ); } - /* - For a given entityRef: - - Fetch entity - - recursively select from refresh_state_references where target is our entityRef. - Continue until we find a location. - - Process and run addUnprocessedEntities with a flag telling it to bump the timestamp for all deferred entities. - */ - /* - For a given URL - - Fetch entity based on managed by location URL? - - repeat for entityRef - */ + const rows = await tx( + 'refresh_state_references', + ) + .where({ target_entity_ref: currentRef }) + .select(); + + if (rows.length === 0) { + if (depth === 1) { + throw new NotFoundError(`Entity ${currentRef} not found`); + } + throw new NotFoundError( + `Entity ${entityRef} has a broken parent reference chain at ${currentRef}`, + ); + } + + const parentRef = rows[0].source_entity_ref; + if (!parentRef) { + // We've reached the top of the tree which is the entityProvider. + // In this case we refresh the entity itself. + break; + } + if (parentRef.startsWith('location:')) { + refreshTarget = parentRef; + break; + } + currentRef = parentRef; } - if ('locationRef' in options) { - // TODO(jhaals): managed by or origin? - // const entity: Entity = JSON.parse(result.processed_entity); - // // TODO: ONly required for URLz - // const managedBy = entity.metadata?.annotations?.[LOCATION_ANNOTATION] - // console.log(managedBy); + const updateResult = await tx('refresh_state') + .where({ entity_ref: refreshTarget }) + .update({ next_update_at: tx.fn.now() }); + if (updateResult === 0) { + throw new ConflictError( + `Failed to schedule ${refreshTarget} for refresh`, + ); } } From d37799f56b01aef374fdecc664f40005603041d0 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Mon, 13 Sep 2021 14:59:47 +0200 Subject: [PATCH 08/23] catalog-backend: Add refresh endpoint to createRouter Co-authored-by: Patrik Oldsberg Signed-off-by: Johan Haals --- plugins/catalog-backend/src/service/router.ts | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/plugins/catalog-backend/src/service/router.ts b/plugins/catalog-backend/src/service/router.ts index 71b8614a99..6a521b4475 100644 --- a/plugins/catalog-backend/src/service/router.ts +++ b/plugins/catalog-backend/src/service/router.ts @@ -28,7 +28,11 @@ import { Logger } from 'winston'; import yn from 'yn'; import { EntitiesCatalog, LocationsCatalog } from '../catalog'; import { HigherOrderOperation, LocationAnalyzer } from '../ingestion/types'; -import { LocationService } from '../next/types'; +import { + LocationService, + CatalogProcessingEngine, + EntityRefreshOptions, +} from '../next/types'; import { basicEntityFilter, parseEntityFilterParams, @@ -47,6 +51,7 @@ export interface RouterOptions { higherOrderOperation?: HigherOrderOperation; locationAnalyzer?: LocationAnalyzer; locationService?: LocationService; + processingEngine?: CatalogProcessingEngine; logger: Logger; config: Config; } @@ -60,6 +65,7 @@ export async function createRouter( higherOrderOperation, locationAnalyzer, locationService, + processingEngine, config, logger, } = options; @@ -73,6 +79,14 @@ export async function createRouter( logger.info('Catalog is running in readonly mode'); } + if (processingEngine) { + router.post('/refresh', async (req, res) => { + const refreshOptions: EntityRefreshOptions = req.body; + await processingEngine.refresh(refreshOptions); + res.status(200).send(); + }); + } + if (entitiesCatalog) { router .get('/entities', async (req, res) => { From 5c8c5ae1e4f88520593ed3d28c773d46b67f3160 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Mon, 13 Sep 2021 15:01:43 +0200 Subject: [PATCH 09/23] catalog-backend: Add documentation Co-authored-by: Patrik Oldsberg Signed-off-by: Johan Haals --- .../src/next/database/DefaultProcessingDatabase.ts | 9 +++++++-- plugins/catalog-backend/src/next/database/types.ts | 11 ++++++++++- 2 files changed, 17 insertions(+), 3 deletions(-) diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index 3cc1ef8233..3ca8fdb10d 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -103,8 +103,6 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { sourceEntityRef: stringifyEntityRef(processedEntity), }); - // Update fragments - // Delete old relations await tx('relations') .where({ originating_entity_id: id }) @@ -329,6 +327,10 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { } } + /** + * Add a set of deferred entities for processing. + * The entities will be added at the front of the processing queue. + */ async addUnprocessedEntities( txOpaque: Transaction, options: AddUnprocessedEntitiesOptions, @@ -352,6 +354,9 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { unprocessed_entity: serializedEntity, location_key: locationKey, last_discovery_at: tx.fn.now(), + // We only get to this point if a processed entity actually had any changes, or + // if an entity provider requested this mutation, meaning that we can safely + // bump the deferred entities to the front of the queue for immediate processing. next_update_at: tx.fn.now(), }) .where('entity_ref', entityRef) diff --git a/plugins/catalog-backend/src/next/database/types.ts b/plugins/catalog-backend/src/next/database/types.ts index 509d9d0d0e..f3e70ea260 100644 --- a/plugins/catalog-backend/src/next/database/types.ts +++ b/plugins/catalog-backend/src/next/database/types.ts @@ -86,6 +86,9 @@ export type RefreshOptions = { export interface ProcessingDatabase { transaction(fn: (tx: Transaction) => Promise): Promise; + /** + * Add unprocessed entities to the front of the processing queue using a mutation. + */ replaceUnprocessedEntities( txOpaque: Transaction, options: ReplaceUnprocessedEntitiesOptions, @@ -97,7 +100,10 @@ export interface ProcessingDatabase { ): Promise; /** - * Updates a processed entity + * Updates a processed entity. + * + * Any deferred entities are added at the front of the processing queue for + * immediate processing, meaning this should only be called when the entity has changes. */ updateProcessedEntity( txOpaque: Transaction, @@ -112,5 +118,8 @@ export interface ProcessingDatabase { options: UpdateProcessedEntityErrorsOptions, ): Promise; + /** + * Schedules a refresh of a given entityRef. + */ refresh(txOpaque: Transaction, options: RefreshOptions): Promise; } From 76ebabc730ab1371922c89f3b3e3724cfe380cab Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Mon, 13 Sep 2021 15:21:51 +0200 Subject: [PATCH 10/23] Add listAncestors method, simplify refresh Co-authored-by: Patrik Oldsberg Signed-off-by: Johan Haals --- .../next/DefaultCatalogProcessingEngine.ts | 11 +++++- .../database/DefaultProcessingDatabase.ts | 34 +++++++++++-------- .../src/next/database/types.ts | 18 ++++++++++ 3 files changed, 47 insertions(+), 16 deletions(-) diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index 6e3f6589d2..998acb230a 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -242,7 +242,16 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { async refresh(options: EntityRefreshOptions) { await this.processingDatabase.transaction(async tx => { - await this.processingDatabase.refresh(tx, options); + const { entityRefs } = await this.processingDatabase.listAncestors(tx, { + entityRef: options.entityRef, + }); + const locationAncestor = entityRefs.find(ref => + ref.startsWith('location:'), + ); + + await this.processingDatabase.refresh(tx, { + entityRef: locationAncestor ?? options.entityRef, + }); }); } } diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index 3ca8fdb10d..a7a04c5f0b 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -39,6 +39,8 @@ import { RefreshOptions, ReplaceUnprocessedEntitiesOptions, UpdateProcessedEntityOptions, + ListAncestorsOptions, + ListAncestorsResult, } from './types'; // The number of items that are sent per batch to the database layer, when @@ -46,7 +48,7 @@ import { // errors in the underlying engine due to exceeding query limits, but large // enough to get the speed benefits. const BATCH_SIZE = 50; -const MAX_REFRESH_DEPTH = 10; +const MAX_ANCESTOR_DEPTH = 32; export class DefaultProcessingDatabase implements ProcessingDatabase { constructor( @@ -516,18 +518,20 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { }; } - async refresh(txOpaque: Transaction, options: RefreshOptions): Promise { + async listAncestors( + txOpaque: Transaction, + options: ListAncestorsOptions, + ): Promise { const tx = txOpaque as Knex.Transaction; const { entityRef } = options; - - let refreshTarget = entityRef; + const entityRefs = new Array(); let currentRef = entityRef; let depth = 0; for (;;) { - if (depth++ > MAX_REFRESH_DEPTH) { + if (depth++ > MAX_ANCESTOR_DEPTH) { throw new Error( - `Unable to refresh Entity ${entityRef}, maximum refresh depth of ${MAX_REFRESH_DEPTH} reached`, + `Unable receive ancestors for ${entityRef}, reached maximum depth of ${MAX_ANCESTOR_DEPTH}`, ); } @@ -550,22 +554,22 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { if (!parentRef) { // We've reached the top of the tree which is the entityProvider. // In this case we refresh the entity itself. - break; - } - if (parentRef.startsWith('location:')) { - refreshTarget = parentRef; - break; + return { entityRefs }; } + entityRefs.push(parentRef); currentRef = parentRef; } + } + + async refresh(txOpaque: Transaction, options: RefreshOptions): Promise { + const tx = txOpaque as Knex.Transaction; + const { entityRef } = options; const updateResult = await tx('refresh_state') - .where({ entity_ref: refreshTarget }) + .where({ entity_ref: entityRef }) .update({ next_update_at: tx.fn.now() }); if (updateResult === 0) { - throw new ConflictError( - `Failed to schedule ${refreshTarget} for refresh`, - ); + throw new ConflictError(`Failed to schedule ${entityRef} for refresh`); } } diff --git a/plugins/catalog-backend/src/next/database/types.ts b/plugins/catalog-backend/src/next/database/types.ts index f3e70ea260..3dfe8e5592 100644 --- a/plugins/catalog-backend/src/next/database/types.ts +++ b/plugins/catalog-backend/src/next/database/types.ts @@ -83,6 +83,14 @@ export type RefreshOptions = { entityRef: string; }; +export type ListAncestorsOptions = { + entityRef: string; +}; + +export type ListAncestorsResult = { + entityRefs: string[]; +}; + export interface ProcessingDatabase { transaction(fn: (tx: Transaction) => Promise): Promise; @@ -122,4 +130,14 @@ export interface ProcessingDatabase { * Schedules a refresh of a given entityRef. */ refresh(txOpaque: Transaction, options: RefreshOptions): Promise; + + /** + * Lists all ancestors of a given entityRef. + * + * The returned list is ordered from the most immediate ancestor to the most distant one. + */ + listAncestors( + txOpaque: Transaction, + options: ListAncestorsOptions, + ): Promise; } From c45976d0f14619f69b9d92896749e90d26cac51e Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Mon, 13 Sep 2021 15:22:28 +0200 Subject: [PATCH 11/23] chore: use timestampToDateTime in test Signed-off-by: Johan Haals --- .../database/DefaultProcessingDatabase.test.ts | 18 ++---------------- 1 file changed, 2 insertions(+), 16 deletions(-) diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts index 23829bd33f..afc3806d36 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts @@ -30,6 +30,7 @@ import { DbRelationsRow, } from './tables'; import { createRandomRefreshInterval } from '../refresh'; +import { timestampToDateTime } from './conversion'; describe('Default Processing Database', () => { const defaultLogger = getVoidLogger(); @@ -66,21 +67,6 @@ describe('Default Processing Database', () => { await db('refresh_state').insert(ref); }; - const parseDate = (date: string | Date): DateTime => { - const parsedDate = - typeof date === 'string' - ? DateTime.fromSQL(date, { zone: 'UTC' }) - : DateTime.fromJSDate(date); - - if (!parsedDate.isValid) { - throw new Error( - `Failed to parse date, reason: ${parsedDate.invalidReason}, explanation: ${parsedDate.invalidExplanation}`, - ); - } - - return parsedDate; - }; - describe('addUprocessedEntities', () => { function mockEntity(name: string, type: string): Entity { return { @@ -1010,7 +996,7 @@ describe('Default Processing Database', () => { const result = await knex('refresh_state') .where('entity_ref', 'location:default/new-root') .select(); - const nextUpdate = parseDate(result[0].next_update_at); + const nextUpdate = timestampToDateTime(result[0].next_update_at); const nextUpdateDiff = nextUpdate.diff(now, 'seconds'); expect(nextUpdateDiff.seconds).toBeGreaterThanOrEqual(90); }, From 89fd81a1abb05b143697805654cf0eb3bd57e24a Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Mon, 13 Sep 2021 16:55:51 +0200 Subject: [PATCH 12/23] Add changesets Signed-off-by: Johan Haals --- .changeset/neat-coats-sell.md | 37 ++++++++++++++++++++++++++++++++ .changeset/red-lizards-accept.md | 5 +++++ 2 files changed, 42 insertions(+) create mode 100644 .changeset/neat-coats-sell.md create mode 100644 .changeset/red-lizards-accept.md diff --git a/.changeset/neat-coats-sell.md b/.changeset/neat-coats-sell.md new file mode 100644 index 0000000000..fd585425be --- /dev/null +++ b/.changeset/neat-coats-sell.md @@ -0,0 +1,37 @@ +--- +'@backstage/create-app': patch +--- + +This change adds an API endpoint for requesting a catalog refresh at `/refresh`, which is activated if a `CatalogProcessingEngine` is passed to `createRouter`. +The creation of the router has been abstracted behind the `CatalogBuilder` to simplify usage and future changes. The following changes are **changes are required** to your `catalog.ts`. + +```diff +- import { +- CatalogBuilder, +- createRouter, +- } from '@backstage/plugin-catalog-backend'; ++ import { CatalogBuilder } from '@backstage/plugin-catalog-backend'; + + export default async function createPlugin( + env: PluginEnvironment, + ): Promise { + const builder = await CatalogBuilder.create(env); +- const { +- entitiesCatalog, +- locationAnalyzer, +- processingEngine, +- locationService, +- } = await builder.build(); ++ const { processingEngine, router } = await builder.build(); + await processingEngine.start(); + +- return await createRouter({ +- entitiesCatalog, +- locationAnalyzer, +- locationService, +- logger: env.logger, +- config: env.config, +- }); ++ return router; + } +``` diff --git a/.changeset/red-lizards-accept.md b/.changeset/red-lizards-accept.md new file mode 100644 index 0000000000..758eb11da5 --- /dev/null +++ b/.changeset/red-lizards-accept.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend': minor +--- + +Add API endpoint for requesting a catalog refresh at `/refresh`, which is activated if a `CatalogProcessingEngine` is passed to `createRouter`. From cf0ac7a900ed7e5e0c53f9ce5707ba63a50c95ed Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Tue, 14 Sep 2021 10:36:35 +0200 Subject: [PATCH 13/23] catalog-backend: export and rename refreshOptions Co-authored-by: Patrik Oldsberg Signed-off-by: Johan Haals --- plugins/catalog-backend/api-report.md | 13 +++++++++++++ .../src/next/DefaultCatalogProcessingEngine.ts | 4 ++-- plugins/catalog-backend/src/next/NextRouter.ts | 4 ++-- plugins/catalog-backend/src/next/index.ts | 1 + plugins/catalog-backend/src/next/types.ts | 5 +++-- plugins/catalog-backend/src/service/router.ts | 4 ++-- 6 files changed, 23 insertions(+), 8 deletions(-) diff --git a/plugins/catalog-backend/api-report.md b/plugins/catalog-backend/api-report.md index 9d542c674d..44073d19e0 100644 --- a/plugins/catalog-backend/api-report.md +++ b/plugins/catalog-backend/api-report.md @@ -27,6 +27,7 @@ import { Organizations } from 'aws-sdk'; import { PluginDatabaseManager } from '@backstage/backend-common'; import { PluginEndpointDiscovery } from '@backstage/backend-common'; import { ResourceEntityV1alpha1 } from '@backstage/catalog-model'; +import { Router } from 'express'; import { ScmIntegrationRegistry } from '@backstage/integration'; import { ScmIntegrations } from '@backstage/integration'; import { UrlReader } from '@backstage/backend-common'; @@ -269,12 +270,19 @@ export type CatalogEnvironment = { // // @public (undocumented) export interface CatalogProcessingEngine { + // (undocumented) + refresh(options: CatalogProcessingEngineRefreshOptions): Promise; // (undocumented) start(): Promise; // (undocumented) stop(): Promise; } +// @public (undocumented) +export type CatalogProcessingEngineRefreshOptions = { + entityRef: string; +}; + // Warning: (ae-missing-release-tag) "CatalogProcessingOrchestrator" is exported by the package, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -1194,6 +1202,7 @@ export class NextCatalogBuilder { locationAnalyzer: LocationAnalyzer; processingEngine: CatalogProcessingEngine; locationService: LocationService; + router: Router; }>; // Warning: (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen replaceEntityPolicies(policies: EntityPolicy[]): NextCatalogBuilder; @@ -1229,6 +1238,8 @@ export interface NextRouterOptions { locationService: LocationService; // (undocumented) logger: Logger_2; + // (undocumented) + processingEngine?: CatalogProcessingEngine; } // Warning: (ae-missing-release-tag) "notFoundError" is exported by the package, but it is missing a release tag (@alpha, @beta, @public, or @internal) @@ -1384,6 +1395,8 @@ export interface RouterOptions { locationService?: LocationService; // (undocumented) logger: Logger_2; + // (undocumented) + processingEngine?: CatalogProcessingEngine; } // Warning: (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index 998acb230a..cb3a9c6e30 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -36,7 +36,7 @@ import { EntityProvider, EntityProviderConnection, EntityProviderMutation, - EntityRefreshOptions, + CatalogProcessingEngineRefreshOptions, } from './types'; class Connection implements EntityProviderConnection { @@ -240,7 +240,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { } } - async refresh(options: EntityRefreshOptions) { + async refresh(options: CatalogProcessingEngineRefreshOptions) { await this.processingDatabase.transaction(async tx => { const { entityRefs } = await this.processingDatabase.listAncestors(tx, { entityRef: options.entityRef, diff --git a/plugins/catalog-backend/src/next/NextRouter.ts b/plugins/catalog-backend/src/next/NextRouter.ts index 03d4005769..4c8a43b10b 100644 --- a/plugins/catalog-backend/src/next/NextRouter.ts +++ b/plugins/catalog-backend/src/next/NextRouter.ts @@ -37,7 +37,7 @@ import { disallowReadonlyMode, validateRequestBody } from '../service/util'; import { CatalogProcessingEngine, LocationService, - EntityRefreshOptions, + CatalogProcessingEngineRefreshOptions, } from './types'; export interface NextRouterOptions { @@ -72,7 +72,7 @@ export async function createNextRouter( if (processingEngine) { router.post('/refresh', async (req, res) => { - const refreshOptions: EntityRefreshOptions = req.body; + const refreshOptions: CatalogProcessingEngineRefreshOptions = req.body; await processingEngine.refresh(refreshOptions); res.status(200).send(); }); diff --git a/plugins/catalog-backend/src/next/index.ts b/plugins/catalog-backend/src/next/index.ts index cdd81d9f4d..b6c973c65d 100644 --- a/plugins/catalog-backend/src/next/index.ts +++ b/plugins/catalog-backend/src/next/index.ts @@ -28,4 +28,5 @@ export type { CatalogProcessingEngine, LocationService, LocationStore, + CatalogProcessingEngineRefreshOptions, } from './types'; diff --git a/plugins/catalog-backend/src/next/types.ts b/plugins/catalog-backend/src/next/types.ts index 91c8c47fe4..6c77dab90c 100644 --- a/plugins/catalog-backend/src/next/types.ts +++ b/plugins/catalog-backend/src/next/types.ts @@ -37,10 +37,11 @@ export interface LocationStore { export interface CatalogProcessingEngine { start(): Promise; stop(): Promise; - refresh(options: EntityRefreshOptions): Promise; + refresh(options: CatalogProcessingEngineRefreshOptions): Promise; } -export type EntityRefreshOptions = { entityRef: string }; +/** @public */ +export type CatalogProcessingEngineRefreshOptions = { entityRef: string }; export type EntityProviderMutation = | { type: 'full'; entities: DeferredEntity[] } diff --git a/plugins/catalog-backend/src/service/router.ts b/plugins/catalog-backend/src/service/router.ts index 6a521b4475..144dc459bb 100644 --- a/plugins/catalog-backend/src/service/router.ts +++ b/plugins/catalog-backend/src/service/router.ts @@ -31,7 +31,7 @@ import { HigherOrderOperation, LocationAnalyzer } from '../ingestion/types'; import { LocationService, CatalogProcessingEngine, - EntityRefreshOptions, + CatalogProcessingEngineRefreshOptions, } from '../next/types'; import { basicEntityFilter, @@ -81,7 +81,7 @@ export async function createRouter( if (processingEngine) { router.post('/refresh', async (req, res) => { - const refreshOptions: EntityRefreshOptions = req.body; + const refreshOptions: CatalogProcessingEngineRefreshOptions = req.body; await processingEngine.refresh(refreshOptions); res.status(200).send(); }); From ddb8152c00802d3001811671c075513cd61a8abb Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Tue, 14 Sep 2021 10:47:08 +0200 Subject: [PATCH 14/23] chore: fix changeset wording Signed-off-by: Johan Haals --- .changeset/neat-coats-sell.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/neat-coats-sell.md b/.changeset/neat-coats-sell.md index fd585425be..8021e8ec18 100644 --- a/.changeset/neat-coats-sell.md +++ b/.changeset/neat-coats-sell.md @@ -3,7 +3,7 @@ --- This change adds an API endpoint for requesting a catalog refresh at `/refresh`, which is activated if a `CatalogProcessingEngine` is passed to `createRouter`. -The creation of the router has been abstracted behind the `CatalogBuilder` to simplify usage and future changes. The following changes are **changes are required** to your `catalog.ts`. +The creation of the router has been abstracted behind the `CatalogBuilder` to simplify usage and future changes. The following **changes are required** to your `catalog.ts` for the refresh endpoint to function. ```diff - import { From 7b40b4512668836f59dc7b5ff31072e3407b9349 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Tue, 14 Sep 2021 10:52:10 +0200 Subject: [PATCH 15/23] chore: Add info about CatalogProcessingEngine Co-authored-by: Patrik Oldsberg Signed-off-by: Johan Haals --- .changeset/red-lizards-accept.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.changeset/red-lizards-accept.md b/.changeset/red-lizards-accept.md index 758eb11da5..9314a35f7f 100644 --- a/.changeset/red-lizards-accept.md +++ b/.changeset/red-lizards-accept.md @@ -3,3 +3,5 @@ --- Add API endpoint for requesting a catalog refresh at `/refresh`, which is activated if a `CatalogProcessingEngine` is passed to `createRouter`. + +The `CatalogProcessingEngine` interface has also received a new `refresh` method, meaning this is a breaking change if you have a custom implementation of it. The new method is used to trigger a refresh of an entity in an as localized was as possible, usually by refreshing the parent location. From 57d462d5ed0f8d5e318d4a12c8e8e936fbb5a8f0 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Thu, 16 Sep 2021 09:31:14 +0200 Subject: [PATCH 16/23] Refactor refreshes into RefreshService MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Patrik Oldsberg Co-authored-by: Fredrik Adelöw Signed-off-by: Johan Haals --- .../DefaultCatalogProcessingEngine.test.ts | 242 +------------ .../next/DefaultCatalogProcessingEngine.ts | 16 - .../src/next/DefaultRefreshService.test.ts | 328 ++++++++++++++++++ .../src/next/DefaultRefreshService.ts | 46 +++ .../src/next/NextCatalogBuilder.ts | 7 +- .../catalog-backend/src/next/NextRouter.ts | 16 +- .../database/DefaultProcessingDatabase.ts | 2 +- plugins/catalog-backend/src/next/index.ts | 2 +- plugins/catalog-backend/src/next/types.ts | 24 +- plugins/catalog-backend/src/service/router.ts | 16 +- 10 files changed, 416 insertions(+), 283 deletions(-) create mode 100644 plugins/catalog-backend/src/next/DefaultRefreshService.test.ts create mode 100644 plugins/catalog-backend/src/next/DefaultRefreshService.ts diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts index a0d62c8642..f7dede4adb 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts @@ -15,27 +15,13 @@ */ import { getVoidLogger } from '@backstage/backend-common'; -import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils'; -import { createHash, Hash } from 'crypto'; -import { Knex } from 'knex'; -import { Logger } from 'winston'; +import { Hash } from 'crypto'; import { DateTime } from 'luxon'; -import { DatabaseManager } from './database/DatabaseManager'; import waitForExpect from 'wait-for-expect'; import { DefaultProcessingDatabase } from './database/DefaultProcessingDatabase'; -import { - DbRefreshStateReferencesRow, - DbRefreshStateRow, -} from './database/tables'; -import { ProcessingDatabase } from './database/types'; import { DefaultCatalogProcessingEngine } from './DefaultCatalogProcessingEngine'; -import { - CatalogProcessingOrchestrator, - EntityProcessingRequest, -} from './processing/types'; +import { CatalogProcessingOrchestrator } from './processing/types'; import { Stitcher } from './stitching/Stitcher'; -import { Entity, stringifyEntityRef } from '@backstage/catalog-model'; -import { v4 as uuid } from 'uuid'; describe('DefaultCatalogProcessingEngine', () => { const db = { @@ -250,227 +236,3 @@ describe('DefaultCatalogProcessingEngine', () => { await engine.stop(); }); }); - -describe('DefaultCatalogProcessingEngine integration', () => { - const defaultLogger = getVoidLogger(); - const databases = TestDatabases.create({ - ids: ['POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'], - }); - - async function createDatabase( - databaseId: TestDatabaseId, - logger: Logger = defaultLogger, - ) { - const knex = await databases.init(databaseId); - await DatabaseManager.createDatabase(knex); - return { - knex, - db: new DefaultProcessingDatabase({ - database: knex, - logger, - refreshInterval: () => 100, - }), - }; - } - - const createPopulatedEngine = async (options: { - db: ProcessingDatabase; - knex: Knex; - entities: Entity[]; - references: { [source: string]: string[] }; - }) => { - const { db, knex, entities, references } = options; - - const entityMap = new Map( - entities.map(entity => [stringifyEntityRef(entity), entity]), - ); - - for (const entity of entities) { - await knex('refresh_state').insert({ - entity_id: uuid(), - entity_ref: stringifyEntityRef(entity), - unprocessed_entity: JSON.stringify(entity), - errors: '[]', - next_update_at: '2031-01-01 23:00:00', - last_discovery_at: '2021-04-01 13:37:00', - }); - } - - for (const entityRef of entityMap.keys()) { - if (!(entityRef in references)) { - await knex( - 'refresh_state_references', - ).insert({ - source_key: 'ConfigLocationProvider', - target_entity_ref: entityRef, - }); - } - } - for (const [sourceRef, targetRefs] of Object.entries(references)) { - for (const targetRef of targetRefs) { - await knex( - 'refresh_state_references', - ).insert({ - source_entity_ref: sourceRef, - target_entity_ref: targetRef, - }); - } - } - - const engine = new DefaultCatalogProcessingEngine( - defaultLogger, - [], - db, - { - async process(request: EntityProcessingRequest) { - const entityRef = stringifyEntityRef(request.entity); - const entity = entityMap.get(entityRef); - if (!entity) { - throw new Error(`Unexpected entity: ${entityRef}`); - } - const deferredEntities = - references[entityRef]?.map(ref => { - const e = entityMap.get(ref); - if (!e) { - throw new Error(`Target entity not found: ${ref}`); - } - return { entity: e, locationKey: ref }; - }) || []; - - return { - ok: true, - completedEntity: { - ...entity, - metadata: { - ...entity.metadata, - annotations: { - ...entity.metadata.annotations, - 'refresh-completed': 'true', - }, - }, - }, - relations: [], - errors: [], - deferredEntities, - state: new Map(), - }; - }, - }, - new Stitcher(knex, defaultLogger), - () => createHash('sha1'), - 50, - ); - - return engine; - }; - - const waitForRefresh = async (knex: Knex, entityRef: string) => { - for (;;) { - const [result] = await knex('refresh_state') - .where('entity_ref', entityRef) - .select(); - - const entity = result.processed_entity - ? (JSON.parse(result.processed_entity) as Entity) - : undefined; - if (entity?.metadata?.annotations?.['refresh-completed']) { - return true; - } - await new Promise(resolve => setTimeout(resolve, 500)); - } - }; - - it.each(databases.eachSupportedId())( - 'should refresh the parent location, %p', - async databaseId => { - const { knex, db } = await createDatabase(databaseId); - - const engine = await createPopulatedEngine({ - db, - knex, - entities: [ - { - kind: 'Location', - apiVersion: '1.0.0', - metadata: { - name: 'myloc', - }, - }, - { - kind: 'Component', - apiVersion: '1.0.0', - metadata: { - name: 'mycomp', - }, - }, - ], - references: { - 'location:default/myloc': ['component:default/mycomp'], - }, - }); - - await engine.start(); - - await engine.refresh({ - entityRef: 'component:default/mycomp', - }); - - await expect( - waitForRefresh(knex, 'component:default/mycomp'), - ).resolves.toBe(true); - - await engine.stop(); - }, - ); - - it.each(databases.eachSupportedId())( - 'should refresh the location further up the tree, %p', - async databaseId => { - const { knex, db } = await createDatabase(databaseId); - - const engine = await createPopulatedEngine({ - db, - knex, - entities: [ - { - kind: 'Location', - apiVersion: '1.0.0', - metadata: { - name: 'myloc', - }, - }, - { - kind: 'Component', - apiVersion: '1.0.0', - metadata: { - name: 'mycomp', - }, - }, - { - kind: 'Api', - apiVersion: '1.0.0', - metadata: { - name: 'myapi', - }, - }, - ], - references: { - 'location:default/myloc': ['component:default/mycomp'], - 'component:default/mycomp': ['api:default/myapi'], - }, - }); - - await engine.start(); - - await engine.refresh({ - entityRef: 'api:default/myapi', - }); - - await expect(waitForRefresh(knex, 'api:default/myapi')).resolves.toBe( - true, - ); - - await engine.stop(); - }, - ); -}); diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index cb3a9c6e30..4d2d74f05b 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -36,7 +36,6 @@ import { EntityProvider, EntityProviderConnection, EntityProviderMutation, - CatalogProcessingEngineRefreshOptions, } from './types'; class Connection implements EntityProviderConnection { @@ -239,21 +238,6 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { this.stopFunc = undefined; } } - - async refresh(options: CatalogProcessingEngineRefreshOptions) { - await this.processingDatabase.transaction(async tx => { - const { entityRefs } = await this.processingDatabase.listAncestors(tx, { - entityRef: options.entityRef, - }); - const locationAncestor = entityRefs.find(ref => - ref.startsWith('location:'), - ); - - await this.processingDatabase.refresh(tx, { - entityRef: locationAncestor ?? options.entityRef, - }); - }); - } } // Helps wrap the timing and logging behaviors diff --git a/plugins/catalog-backend/src/next/DefaultRefreshService.test.ts b/plugins/catalog-backend/src/next/DefaultRefreshService.test.ts new file mode 100644 index 0000000000..5df57c6897 --- /dev/null +++ b/plugins/catalog-backend/src/next/DefaultRefreshService.test.ts @@ -0,0 +1,328 @@ +/* + * Copyright 2021 The Backstage Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { getVoidLogger } from '@backstage/backend-common'; +import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils'; +import { createHash } from 'crypto'; +import { Knex } from 'knex'; +import { Logger } from 'winston'; +import { DatabaseManager } from './database/DatabaseManager'; +import { DefaultProcessingDatabase } from './database/DefaultProcessingDatabase'; +import { + DbRefreshStateReferencesRow, + DbRefreshStateRow, +} from './database/tables'; +import { ProcessingDatabase } from './database/types'; +import { DefaultCatalogProcessingEngine } from './DefaultCatalogProcessingEngine'; +import { EntityProcessingRequest } from './processing/types'; +import { Stitcher } from './stitching/Stitcher'; +import { Entity, stringifyEntityRef } from '@backstage/catalog-model'; +import { v4 as uuid } from 'uuid'; +import { DefaultRefreshService } from './DefaultRefreshService'; + +describe('Refresh integration', () => { + const defaultLogger = getVoidLogger(); + const databases = TestDatabases.create({ + ids: ['POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'], + }); + + async function createDatabase( + databaseId: TestDatabaseId, + logger: Logger = defaultLogger, + ) { + const knex = await databases.init(databaseId); + await DatabaseManager.createDatabase(knex); + return { + knex, + db: new DefaultProcessingDatabase({ + database: knex, + logger, + refreshInterval: () => 100, + }), + }; + } + + const createPopulatedEngine = async (options: { + db: ProcessingDatabase; + knex: Knex; + entities: Entity[]; + references: { [source: string]: string[] }; + entityProcessor?: (entity: Entity) => void; + }) => { + const { db, knex, entities, references, entityProcessor } = options; + + const entityMap = new Map( + entities.map(entity => [stringifyEntityRef(entity), entity]), + ); + + for (const entity of entities) { + await knex('refresh_state').insert({ + entity_id: uuid(), + entity_ref: stringifyEntityRef(entity), + unprocessed_entity: JSON.stringify(entity), + errors: '[]', + next_update_at: '2031-01-01 23:00:00', + last_discovery_at: '2021-04-01 13:37:00', + }); + } + + const entitiesWithParent = new Set(Object.values(references).flat()); + for (const entityRef of entityMap.keys()) { + if (!entitiesWithParent.has(entityRef)) { + await knex( + 'refresh_state_references', + ).insert({ + source_key: 'ConfigLocationProvider', + target_entity_ref: entityRef, + }); + } + } + for (const [sourceRef, targetRefs] of Object.entries(references)) { + for (const targetRef of targetRefs) { + await knex( + 'refresh_state_references', + ).insert({ + source_entity_ref: sourceRef, + target_entity_ref: targetRef, + }); + } + } + + const engine = new DefaultCatalogProcessingEngine( + defaultLogger, + [], + db, + { + async process(request: EntityProcessingRequest) { + const entityRef = stringifyEntityRef(request.entity); + const entity = entityMap.get(entityRef); + if (!entity) { + throw new Error(`Unexpected entity: ${entityRef}`); + } + const deferredEntities = + references[entityRef]?.map(ref => { + const e = entityMap.get(ref); + if (!e) { + throw new Error(`Target entity not found: ${ref}`); + } + return { entity: e, locationKey: ref }; + }) || []; + + entityProcessor?.(entity); + + return { + ok: true, + completedEntity: { + ...entity, + metadata: { + ...entity.metadata, + annotations: { + ...entity.metadata.annotations, + 'refresh-completed': 'true', + }, + }, + }, + relations: [], + errors: [], + deferredEntities, + state: new Map(), + }; + }, + }, + new Stitcher(knex, defaultLogger), + () => createHash('sha1'), + 50, + ); + + return engine; + }; + + const waitForRefresh = async (knex: Knex, entityRef: string) => { + for (;;) { + const [result] = await knex('refresh_state') + .where('entity_ref', entityRef) + .select(); + + const entity = result.processed_entity + ? (JSON.parse(result.processed_entity) as Entity) + : undefined; + if (entity?.metadata?.annotations?.['refresh-completed']) { + // Reset the annotation so that we can run another verification + delete entity.metadata.annotations['refresh-completed']; + await knex('refresh_state') + .update({ + processed_entity: JSON.stringify(entity), + }) + .where('entity_ref', entityRef); + return true; + } + await new Promise(resolve => setTimeout(resolve, 500)); + } + }; + + it.each(databases.eachSupportedId())( + 'should refresh the parent location, %p', + async databaseId => { + const { knex, db } = await createDatabase(databaseId); + const refreshService = new DefaultRefreshService({ database: db }); + const engine = await createPopulatedEngine({ + db, + knex, + entities: [ + { + kind: 'Location', + apiVersion: '1.0.0', + metadata: { + name: 'myloc', + }, + }, + { + kind: 'Component', + apiVersion: '1.0.0', + metadata: { + name: 'mycomp', + }, + }, + ], + references: { + 'location:default/myloc': ['component:default/mycomp'], + }, + }); + + await engine.start(); + + await refreshService.refresh({ + entityRef: 'component:default/mycomp', + }); + + await expect( + waitForRefresh(knex, 'location:default/myloc'), + ).resolves.toBe(true); + + await engine.stop(); + }, + ); + + it.each(databases.eachSupportedId())( + 'should refresh the location further up the tree, %p', + async databaseId => { + const { knex, db } = await createDatabase(databaseId); + const refreshService = new DefaultRefreshService({ database: db }); + const engine = await createPopulatedEngine({ + db, + knex, + entities: [ + { + kind: 'Location', + apiVersion: '1.0.0', + metadata: { + name: 'myloc', + }, + }, + { + kind: 'Component', + apiVersion: '1.0.0', + metadata: { + name: 'mycomp', + }, + }, + { + kind: 'Api', + apiVersion: '1.0.0', + metadata: { + name: 'myapi', + }, + }, + ], + references: { + 'location:default/myloc': ['component:default/mycomp'], + 'component:default/mycomp': ['api:default/myapi'], + }, + }); + + await engine.start(); + + await refreshService.refresh({ + entityRef: 'api:default/myapi', + }); + + await expect(waitForRefresh(knex, 'api:default/myapi')).resolves.toBe( + true, + ); + + await engine.stop(); + }, + ); + + it.each(databases.eachSupportedId())( + 'should refresh even when parent has no changes', + async databaseId => { + let secondRound = false; + const { knex, db } = await createDatabase(databaseId); + const refreshService = new DefaultRefreshService({ database: db }); + const engine = await createPopulatedEngine({ + db, + knex, + entities: [ + { + kind: 'Location', + apiVersion: '1.0.0', + metadata: { + name: 'myloc', + }, + }, + { + kind: 'Component', + apiVersion: '1.0.0', + metadata: { + name: 'mycomp', + }, + }, + ], + references: { + 'location:default/myloc': ['component:default/mycomp'], + }, + entityProcessor: entity => { + if (entity.metadata.name === 'mycomp' && secondRound) { + entity.apiVersion = '2.0.0'; + } + }, + }); + + await engine.start(); + + await refreshService.refresh({ + entityRef: 'component:default/mycomp', + }); + + await expect( + waitForRefresh(knex, 'component:default/mycomp'), + ).resolves.toBe(true); + + secondRound = true; + + await refreshService.refresh({ + entityRef: 'component:default/mycomp', + }); + + await expect( + waitForRefresh(knex, 'component:default/mycomp'), + ).resolves.toBe(true); + + await engine.stop(); + }, + ); +}); diff --git a/plugins/catalog-backend/src/next/DefaultRefreshService.ts b/plugins/catalog-backend/src/next/DefaultRefreshService.ts new file mode 100644 index 0000000000..bb713fd355 --- /dev/null +++ b/plugins/catalog-backend/src/next/DefaultRefreshService.ts @@ -0,0 +1,46 @@ +/* + * Copyright 2021 The Backstage Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { DefaultProcessingDatabase } from './database/DefaultProcessingDatabase'; +import { RefreshOptions, RefreshService } from './types'; + +export class DefaultRefreshService implements RefreshService { + private database: DefaultProcessingDatabase; + + constructor(options: { database: DefaultProcessingDatabase }) { + this.database = options.database; + } + + async refresh(options: RefreshOptions) { + await this.database.transaction(async tx => { + const { entityRefs } = await this.database.listAncestors(tx, { + entityRef: options.entityRef, + }); + const locationAncestor = entityRefs.find(ref => + ref.startsWith('location:'), + ); + + if (locationAncestor) { + await this.database.refresh(tx, { + entityRef: locationAncestor, + }); + } + await this.database.refresh(tx, { + entityRef: options.entityRef, + }); + }); + } +} diff --git a/plugins/catalog-backend/src/next/NextCatalogBuilder.ts b/plugins/catalog-backend/src/next/NextCatalogBuilder.ts index 9186fc3efe..a34be2ca6a 100644 --- a/plugins/catalog-backend/src/next/NextCatalogBuilder.ts +++ b/plugins/catalog-backend/src/next/NextCatalogBuilder.ts @@ -77,6 +77,7 @@ import { } from './refresh'; import { CatalogEnvironment } from '../service/CatalogBuilder'; import { createNextRouter } from './NextRouter'; +import { DefaultRefreshService } from './DefaultRefreshService'; /** * A builder that helps wire up all of the component parts of the catalog. @@ -335,12 +336,14 @@ export class NextCatalogBuilder { locationStore, orchestrator, ); - + const refreshService = new DefaultRefreshService({ + database: processingDatabase, + }); const router = await createNextRouter({ entitiesCatalog, locationAnalyzer, locationService, - processingEngine, + refreshService, logger, config, }); diff --git a/plugins/catalog-backend/src/next/NextRouter.ts b/plugins/catalog-backend/src/next/NextRouter.ts index 4c8a43b10b..b9da91b30a 100644 --- a/plugins/catalog-backend/src/next/NextRouter.ts +++ b/plugins/catalog-backend/src/next/NextRouter.ts @@ -34,17 +34,13 @@ import { parseEntityTransformParams, } from '../service/request'; import { disallowReadonlyMode, validateRequestBody } from '../service/util'; -import { - CatalogProcessingEngine, - LocationService, - CatalogProcessingEngineRefreshOptions, -} from './types'; +import { RefreshService, RefreshOptions, LocationService } from './types'; export interface NextRouterOptions { entitiesCatalog?: EntitiesCatalog; locationAnalyzer?: LocationAnalyzer; locationService: LocationService; - processingEngine?: CatalogProcessingEngine; + refreshService?: RefreshService; logger: Logger; config: Config; } @@ -56,7 +52,7 @@ export async function createNextRouter( entitiesCatalog, locationAnalyzer, locationService, - processingEngine, + refreshService, config, logger, } = options; @@ -70,10 +66,10 @@ export async function createNextRouter( logger.info('Catalog is running in readonly mode'); } - if (processingEngine) { + if (refreshService) { router.post('/refresh', async (req, res) => { - const refreshOptions: CatalogProcessingEngineRefreshOptions = req.body; - await processingEngine.refresh(refreshOptions); + const refreshOptions: RefreshOptions = req.body; + await refreshService.refresh(refreshOptions); res.status(200).send(); }); } diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index a7a04c5f0b..472c79e516 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -550,7 +550,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { ); } - const parentRef = rows[0].source_entity_ref; + const parentRef = rows.find(r => r.source_entity_ref)?.source_entity_ref; if (!parentRef) { // We've reached the top of the tree which is the entityProvider. // In this case we refresh the entity itself. diff --git a/plugins/catalog-backend/src/next/index.ts b/plugins/catalog-backend/src/next/index.ts index b6c973c65d..f05ba903eb 100644 --- a/plugins/catalog-backend/src/next/index.ts +++ b/plugins/catalog-backend/src/next/index.ts @@ -28,5 +28,5 @@ export type { CatalogProcessingEngine, LocationService, LocationStore, - CatalogProcessingEngineRefreshOptions, + RefreshOptions, } from './types'; diff --git a/plugins/catalog-backend/src/next/types.ts b/plugins/catalog-backend/src/next/types.ts index 6c77dab90c..fa058c4138 100644 --- a/plugins/catalog-backend/src/next/types.ts +++ b/plugins/catalog-backend/src/next/types.ts @@ -37,11 +37,29 @@ export interface LocationStore { export interface CatalogProcessingEngine { start(): Promise; stop(): Promise; - refresh(options: CatalogProcessingEngineRefreshOptions): Promise; } -/** @public */ -export type CatalogProcessingEngineRefreshOptions = { entityRef: string }; +/** + * Options for requesting a refresh of entities in the catalog. + * + * @public + */ +export type RefreshOptions = { + /** The reference to a single entity that should be refreshed */ + entityRef: string; +}; + +/** + * A service that manages refreshes of entities in the catalog. + * + * @public + */ +export interface RefreshService { + /** + * Request a refresh of entities in the catalog. + */ + refresh(options: RefreshOptions): Promise; +} export type EntityProviderMutation = | { type: 'full'; entities: DeferredEntity[] } diff --git a/plugins/catalog-backend/src/service/router.ts b/plugins/catalog-backend/src/service/router.ts index 144dc459bb..f1bb18c825 100644 --- a/plugins/catalog-backend/src/service/router.ts +++ b/plugins/catalog-backend/src/service/router.ts @@ -28,11 +28,7 @@ import { Logger } from 'winston'; import yn from 'yn'; import { EntitiesCatalog, LocationsCatalog } from '../catalog'; import { HigherOrderOperation, LocationAnalyzer } from '../ingestion/types'; -import { - LocationService, - CatalogProcessingEngine, - CatalogProcessingEngineRefreshOptions, -} from '../next/types'; +import { RefreshService, LocationService, RefreshOptions } from '../next/types'; import { basicEntityFilter, parseEntityFilterParams, @@ -51,7 +47,7 @@ export interface RouterOptions { higherOrderOperation?: HigherOrderOperation; locationAnalyzer?: LocationAnalyzer; locationService?: LocationService; - processingEngine?: CatalogProcessingEngine; + refreshService?: RefreshService; logger: Logger; config: Config; } @@ -65,7 +61,7 @@ export async function createRouter( higherOrderOperation, locationAnalyzer, locationService, - processingEngine, + refreshService, config, logger, } = options; @@ -79,10 +75,10 @@ export async function createRouter( logger.info('Catalog is running in readonly mode'); } - if (processingEngine) { + if (refreshService) { router.post('/refresh', async (req, res) => { - const refreshOptions: CatalogProcessingEngineRefreshOptions = req.body; - await processingEngine.refresh(refreshOptions); + const refreshOptions: RefreshOptions = req.body; + await refreshService.refresh(refreshOptions); res.status(200).send(); }); } From 6a208c5e233e719c8c68406b584e2effe20438e9 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Thu, 16 Sep 2021 09:37:31 +0200 Subject: [PATCH 17/23] Reword changesets after refactor Signed-off-by: Johan Haals --- .changeset/neat-coats-sell.md | 2 +- .changeset/red-lizards-accept.md | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/.changeset/neat-coats-sell.md b/.changeset/neat-coats-sell.md index 8021e8ec18..80c9ba111c 100644 --- a/.changeset/neat-coats-sell.md +++ b/.changeset/neat-coats-sell.md @@ -2,7 +2,7 @@ '@backstage/create-app': patch --- -This change adds an API endpoint for requesting a catalog refresh at `/refresh`, which is activated if a `CatalogProcessingEngine` is passed to `createRouter`. +This change adds an API endpoint for requesting a catalog refresh at `/refresh`, which is activated if a `RefreshService` is passed to `createRouter`. The creation of the router has been abstracted behind the `CatalogBuilder` to simplify usage and future changes. The following **changes are required** to your `catalog.ts` for the refresh endpoint to function. ```diff diff --git a/.changeset/red-lizards-accept.md b/.changeset/red-lizards-accept.md index 9314a35f7f..74c3a76db5 100644 --- a/.changeset/red-lizards-accept.md +++ b/.changeset/red-lizards-accept.md @@ -2,6 +2,6 @@ '@backstage/plugin-catalog-backend': minor --- -Add API endpoint for requesting a catalog refresh at `/refresh`, which is activated if a `CatalogProcessingEngine` is passed to `createRouter`. +Add API endpoint for requesting a catalog refresh at `/refresh`, which is activated if a `RefreshService` is passed to `createRouter`. -The `CatalogProcessingEngine` interface has also received a new `refresh` method, meaning this is a breaking change if you have a custom implementation of it. The new method is used to trigger a refresh of an entity in an as localized was as possible, usually by refreshing the parent location. +The new method is used to trigger a refresh of an entity in an as localized was as possible, usually by refreshing the parent location. From 70a58c73286a6d0fb46d4af38554af0bd390e882 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Thu, 16 Sep 2021 10:29:24 +0200 Subject: [PATCH 18/23] Export RefreshService, add api-report Signed-off-by: Johan Haals --- plugins/catalog-backend/api-report.md | 21 ++++++++++++--------- plugins/catalog-backend/src/next/index.ts | 1 + 2 files changed, 13 insertions(+), 9 deletions(-) diff --git a/plugins/catalog-backend/api-report.md b/plugins/catalog-backend/api-report.md index 44073d19e0..76663e3c95 100644 --- a/plugins/catalog-backend/api-report.md +++ b/plugins/catalog-backend/api-report.md @@ -270,19 +270,12 @@ export type CatalogEnvironment = { // // @public (undocumented) export interface CatalogProcessingEngine { - // (undocumented) - refresh(options: CatalogProcessingEngineRefreshOptions): Promise; // (undocumented) start(): Promise; // (undocumented) stop(): Promise; } -// @public (undocumented) -export type CatalogProcessingEngineRefreshOptions = { - entityRef: string; -}; - // Warning: (ae-missing-release-tag) "CatalogProcessingOrchestrator" is exported by the package, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -1239,7 +1232,7 @@ export interface NextRouterOptions { // (undocumented) logger: Logger_2; // (undocumented) - processingEngine?: CatalogProcessingEngine; + refreshService?: RefreshService; } // Warning: (ae-missing-release-tag) "notFoundError" is exported by the package, but it is missing a release tag (@alpha, @beta, @public, or @internal) @@ -1360,6 +1353,16 @@ export type RecursivePartial = { // @public export type RefreshIntervalFunction = () => number; +// @public +export type RefreshOptions = { + entityRef: string; +}; + +// @public +export interface RefreshService { + refresh(options: RefreshOptions): Promise; +} + // Warning: (ae-missing-release-tag) "relation" is exported by the package, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -1396,7 +1399,7 @@ export interface RouterOptions { // (undocumented) logger: Logger_2; // (undocumented) - processingEngine?: CatalogProcessingEngine; + refreshService?: RefreshService; } // Warning: (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen diff --git a/plugins/catalog-backend/src/next/index.ts b/plugins/catalog-backend/src/next/index.ts index f05ba903eb..5f70dd2125 100644 --- a/plugins/catalog-backend/src/next/index.ts +++ b/plugins/catalog-backend/src/next/index.ts @@ -29,4 +29,5 @@ export type { LocationService, LocationStore, RefreshOptions, + RefreshService, } from './types'; From de3db12c21f7b91df8ff9359270163668555a179 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Thu, 16 Sep 2021 11:56:56 +0200 Subject: [PATCH 19/23] Update plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Johan Haals Co-authored-by: Fredrik Adelöw --- .../src/next/database/DefaultProcessingDatabase.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index 472c79e516..c57b5977a7 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -566,7 +566,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { const { entityRef } = options; const updateResult = await tx('refresh_state') - .where({ entity_ref: entityRef }) + .where({ entity_ref: entityRef.toLocaleLowerCase('en-US') }) .update({ next_update_at: tx.fn.now() }); if (updateResult === 0) { throw new ConflictError(`Failed to schedule ${entityRef} for refresh`); From c54ea11bcdaa36cf954cc2fd5bd2a76cf8cba711 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Thu, 16 Sep 2021 11:57:16 +0200 Subject: [PATCH 20/23] Update plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Johan Haals Co-authored-by: Fredrik Adelöw --- .../src/next/database/DefaultProcessingDatabase.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index c57b5977a7..666a2b10f6 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -526,7 +526,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { const { entityRef } = options; const entityRefs = new Array(); - let currentRef = entityRef; + let currentRef = entityRef.toLocaleLowerCase('en-US'); let depth = 0; for (;;) { if (depth++ > MAX_ANCESTOR_DEPTH) { From 80965dcc5801e192491f99b64a0511a89b2af4e9 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Thu, 16 Sep 2021 11:58:27 +0200 Subject: [PATCH 21/23] Update plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Johan Haals Co-authored-by: Fredrik Adelöw --- .../src/next/database/DefaultProcessingDatabase.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index 666a2b10f6..5fc6f96920 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -569,7 +569,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { .where({ entity_ref: entityRef.toLocaleLowerCase('en-US') }) .update({ next_update_at: tx.fn.now() }); if (updateResult === 0) { - throw new ConflictError(`Failed to schedule ${entityRef} for refresh`); + throw new NotFoundError(`Failed to schedule ${entityRef} for refresh`); } } From 9415f38915e8346f105c05c411aa470fe38b1253 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Thu, 16 Sep 2021 12:47:31 +0200 Subject: [PATCH 22/23] chore: Simplify acestor loop Signed-off-by: Johan Haals --- .../src/next/database/DefaultProcessingDatabase.ts | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index 5fc6f96920..8ba9d54b8f 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -527,14 +527,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { const entityRefs = new Array(); let currentRef = entityRef.toLocaleLowerCase('en-US'); - let depth = 0; - for (;;) { - if (depth++ > MAX_ANCESTOR_DEPTH) { - throw new Error( - `Unable receive ancestors for ${entityRef}, reached maximum depth of ${MAX_ANCESTOR_DEPTH}`, - ); - } - + for (let depth = 1; depth <= MAX_ANCESTOR_DEPTH; depth += 1) { const rows = await tx( 'refresh_state_references', ) @@ -559,6 +552,9 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { entityRefs.push(parentRef); currentRef = parentRef; } + throw new Error( + `Unable receive ancestors for ${entityRef}, reached maximum depth of ${MAX_ANCESTOR_DEPTH}`, + ); } async refresh(txOpaque: Transaction, options: RefreshOptions): Promise { From 49fed0947e4cd0b4cca38b259667a47b9592f8b0 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Thu, 16 Sep 2021 12:56:11 +0200 Subject: [PATCH 23/23] chore: Add todo for future improvements Signed-off-by: Johan Haals --- plugins/catalog-backend/src/next/DefaultRefreshService.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/plugins/catalog-backend/src/next/DefaultRefreshService.ts b/plugins/catalog-backend/src/next/DefaultRefreshService.ts index bb713fd355..cfc7fa220b 100644 --- a/plugins/catalog-backend/src/next/DefaultRefreshService.ts +++ b/plugins/catalog-backend/src/next/DefaultRefreshService.ts @@ -33,6 +33,8 @@ export class DefaultRefreshService implements RefreshService { ref.startsWith('location:'), ); + // TODO: Refreshes are currently scheduled(as soon as possible) for execution and will therefore happen in the future. + // There's room for improvements here where the refresh could potentially hang or return an ID so that the user can check progress. if (locationAncestor) { await this.database.refresh(tx, { entityRef: locationAncestor,