From 9890488abbda4b9e35cec5cb4c11be248d34c448 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Mon, 22 Sep 2025 14:58:12 +0200 Subject: [PATCH 1/2] make some inputs mandatory, that are always provided nowadays MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- .changeset/brave-teeth-reply.md | 5 + .../DefaultCatalogProcessingEngine.test.ts | 9 + .../DefaultCatalogProcessingEngine.ts | 30 ++- .../src/service/CatalogBuilder.ts | 4 +- .../src/service/DefaultRefreshService.test.ts | 1 + .../src/service/createRouter.test.ts | 6 +- .../src/service/createRouter.ts | 179 +++++++++--------- .../src/tests/integration.test.ts | 1 + 8 files changed, 120 insertions(+), 115 deletions(-) create mode 100644 .changeset/brave-teeth-reply.md diff --git a/.changeset/brave-teeth-reply.md b/.changeset/brave-teeth-reply.md new file mode 100644 index 0000000000..8a500692a7 --- /dev/null +++ b/.changeset/brave-teeth-reply.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend': patch +--- + +Internal refactor to remove remnants of the old backend system diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts index ab54a7b885..93c22aab1e 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts @@ -70,6 +70,7 @@ describe('DefaultCatalogProcessingEngine', () => { orchestrator: orchestrator, stitcher: stitcher, createHash: () => hash, + scheduler: mockServices.scheduler(), }); db.transaction.mockImplementation(cb => cb((() => {}) as any)); @@ -136,6 +137,7 @@ describe('DefaultCatalogProcessingEngine', () => { knex: {} as any, orchestrator: orchestrator, stitcher: stitcher, + scheduler: mockServices.scheduler(), createHash: () => hash, }); @@ -219,6 +221,7 @@ describe('DefaultCatalogProcessingEngine', () => { knex: {} as any, orchestrator: orchestrator, stitcher: stitcher, + scheduler: mockServices.scheduler(), createHash: () => hash, }); @@ -296,6 +299,7 @@ describe('DefaultCatalogProcessingEngine', () => { knex: {} as any, orchestrator: orchestrator, stitcher: stitcher, + scheduler: mockServices.scheduler(), createHash: () => hash, }); @@ -355,6 +359,7 @@ describe('DefaultCatalogProcessingEngine', () => { knex: {} as any, orchestrator: orchestrator, stitcher: stitcher, + scheduler: mockServices.scheduler(), createHash: () => hash, pollingIntervalMs: 100, }); @@ -470,6 +475,7 @@ describe('DefaultCatalogProcessingEngine', () => { knex: {} as any, orchestrator: orchestrator, stitcher: stitcher, + scheduler: mockServices.scheduler(), createHash: () => hash, pollingIntervalMs: 100, }); @@ -575,6 +581,7 @@ describe('DefaultCatalogProcessingEngine', () => { knex: {} as any, orchestrator: orchestrator, stitcher: stitcher, + scheduler: mockServices.scheduler(), createHash: () => hash, pollingIntervalMs: 100, }); @@ -658,6 +665,7 @@ describe('DefaultCatalogProcessingEngine', () => { knex: {} as any, orchestrator: orchestrator, stitcher: stitcher, + scheduler: mockServices.scheduler(), createHash: () => hash, pollingIntervalMs: 100, }); @@ -746,6 +754,7 @@ describe('DefaultCatalogProcessingEngine', () => { knex: {} as any, orchestrator: orchestrator, stitcher: stitcher, + scheduler: mockServices.scheduler(), createHash: () => hash, pollingIntervalMs: 100, }); diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts index 266b80f9a4..7061309175 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts @@ -59,7 +59,7 @@ const stableStringifyArray = (arr: any[]) => { // is just one. export class DefaultCatalogProcessingEngine { private readonly config: Config; - private readonly scheduler?: SchedulerService; + private readonly scheduler: SchedulerService; private readonly logger: LoggerService; private readonly knex: Knex; private readonly processingDatabase: ProcessingDatabase; @@ -79,7 +79,7 @@ export class DefaultCatalogProcessingEngine { constructor(options: { config: Config; - scheduler?: SchedulerService; + scheduler: SchedulerService; logger: LoggerService; knex: Knex; processingDatabase: ProcessingDatabase; @@ -370,25 +370,17 @@ export class DefaultCatalogProcessingEngine { } }; - if (this.scheduler) { - const abortController = new AbortController(); + const abortController = new AbortController(); + this.scheduler.scheduleTask({ + id: 'catalog_orphan_cleanup', + frequency: { milliseconds: this.orphanCleanupIntervalMs }, + timeout: { milliseconds: this.orphanCleanupIntervalMs * 0.8 }, + fn: runOnce, + signal: abortController.signal, + }); - this.scheduler.scheduleTask({ - id: 'catalog_orphan_cleanup', - frequency: { milliseconds: this.orphanCleanupIntervalMs }, - timeout: { milliseconds: this.orphanCleanupIntervalMs * 0.8 }, - fn: runOnce, - signal: abortController.signal, - }); - - return () => { - abortController.abort(); - }; - } - - const intervalKey = setInterval(runOnce, this.orphanCleanupIntervalMs); return () => { - clearInterval(intervalKey); + abortController.abort(); }; } } diff --git a/plugins/catalog-backend/src/service/CatalogBuilder.ts b/plugins/catalog-backend/src/service/CatalogBuilder.ts index dcd677152e..40fcc3a624 100644 --- a/plugins/catalog-backend/src/service/CatalogBuilder.ts +++ b/plugins/catalog-backend/src/service/CatalogBuilder.ts @@ -119,10 +119,10 @@ export type CatalogEnvironment = { reader: UrlReaderService; permissions: PermissionsService | PermissionAuthorizer; permissionsRegistry?: PermissionsRegistryService; - scheduler?: SchedulerService; + scheduler: SchedulerService; auth: AuthService; httpAuth: HttpAuthService; - auditor?: AuditorService; + auditor: AuditorService; }; /** diff --git a/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts b/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts index a89b21f79f..f999400b8a 100644 --- a/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts +++ b/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts @@ -121,6 +121,7 @@ describe('DefaultRefreshService', () => { processingDatabase: db, knex: knex, stitcher: stitcher, + scheduler: mockServices.scheduler(), orchestrator: { async process(request: EntityProcessingRequest) { const entityRef = stringifyEntityRef(request.entity); diff --git a/plugins/catalog-backend/src/service/createRouter.test.ts b/plugins/catalog-backend/src/service/createRouter.test.ts index 0246a3d56b..299395552a 100644 --- a/plugins/catalog-backend/src/service/createRouter.test.ts +++ b/plugins/catalog-backend/src/service/createRouter.test.ts @@ -156,12 +156,12 @@ describe('createRouter readonly disabled', () => { logger: mockServices.logger.mock(), refreshService, config: new ConfigReader(undefined), - permissionIntegrationRouter: express.Router(), auth: mockServices.auth(), httpAuth: mockServices.httpAuth(), locationAnalyzer, permissionsService, enableRelationsCompatibility: true, // added + auditor: mockServices.auditor.mock(), }); app = await wrapServer(express().use(router)); @@ -218,12 +218,12 @@ describe('createRouter readonly disabled', () => { logger: mockServices.logger.mock(), refreshService, config: new ConfigReader(undefined), - permissionIntegrationRouter: express.Router(), auth: mockServices.auth(), httpAuth: mockServices.httpAuth(), locationAnalyzer, permissionsService, enableRelationsCompatibility: true, + auditor: mockServices.auditor.mock(), }); app = await wrapServer(express().use(router)); entitiesCatalog.entities.mockResolvedValueOnce({ @@ -951,6 +951,7 @@ describe('createRouter readonly and raw json enabled', () => { permissionIntegrationRouter: express.Router(), auth: mockServices.auth(), httpAuth: mockServices.httpAuth(), + orchestrator: { process: jest.fn() }, permissionsService, auditor: mockServices.auditor.mock(), }); @@ -1171,6 +1172,7 @@ describe('NextRouter permissioning', () => { }), auth: mockServices.auth(), httpAuth: mockServices.httpAuth(), + orchestrator: { process: jest.fn() }, permissionsService, auditor: mockServices.auditor.mock(), }); diff --git a/plugins/catalog-backend/src/service/createRouter.ts b/plugins/catalog-backend/src/service/createRouter.ts index 9c9b97eb4c..dc7678a3b9 100644 --- a/plugins/catalog-backend/src/service/createRouter.ts +++ b/plugins/catalog-backend/src/service/createRouter.ts @@ -20,7 +20,6 @@ import { HttpAuthService, LoggerService, PermissionsService, - SchedulerService, } from '@backstage/backend-plugin-api'; import { ANNOTATION_LOCATION, @@ -70,17 +69,15 @@ export interface RouterOptions { entitiesCatalog?: EntitiesCatalog; locationAnalyzer?: LocationAnalyzer; locationService: LocationService; - orchestrator?: CatalogProcessingOrchestrator; + orchestrator: CatalogProcessingOrchestrator; refreshService?: RefreshService; - scheduler?: SchedulerService; logger: LoggerService; config: Config; permissionIntegrationRouter?: express.Router; auth: AuthService; httpAuth: HttpAuthService; permissionsService: PermissionsService; - // TODO: Require AuditorService once `backend-legacy` is removed - auditor?: AuditorService; + auditor: AuditorService; enableRelationsCompatibility?: boolean; } @@ -124,7 +121,7 @@ export async function createRouter( router.post('/refresh', async (req, res) => { const { authorizationToken, ...restBody } = req.body; - const auditorEvent = await auditor?.createEvent({ + const auditorEvent = await auditor.createEvent({ eventId: 'entity-mutate', severityLevel: 'medium', meta: { @@ -160,7 +157,7 @@ export async function createRouter( if (entitiesCatalog) { router .get('/entities', async (req, res) => { - const auditorEvent = await auditor?.createEvent({ + const auditorEvent = await auditor.createEvent({ eventId: 'entity-fetch', request: req, meta: { @@ -258,7 +255,7 @@ export async function createRouter( } }) .get('/entities/by-query', async (req, res) => { - const auditorEvent = await auditor?.createEvent({ + const auditorEvent = await auditor.createEvent({ eventId: 'entity-fetch', request: req, meta: { @@ -311,7 +308,7 @@ export async function createRouter( .get('/entities/by-uid/:uid', async (req, res) => { const { uid } = req.params; - const auditorEvent = await auditor?.createEvent({ + const auditorEvent = await auditor.createEvent({ eventId: 'entity-fetch', request: req, meta: { @@ -356,7 +353,7 @@ export async function createRouter( .delete('/entities/by-uid/:uid', async (req, res) => { const { uid } = req.params; - const auditorEvent = await auditor?.createEvent({ + const auditorEvent = await auditor.createEvent({ eventId: 'entity-mutate', severityLevel: 'medium', request: req, @@ -385,7 +382,7 @@ export async function createRouter( const { kind, namespace, name } = req.params; const entityRef = stringifyEntityRef({ kind, namespace, name }); - const auditorEvent = await auditor?.createEvent({ + const auditorEvent = await auditor.createEvent({ eventId: 'entity-fetch', request: req, meta: { @@ -420,7 +417,7 @@ export async function createRouter( const { kind, namespace, name } = req.params; const entityRef = stringifyEntityRef({ kind, namespace, name }); - const auditorEvent = await auditor?.createEvent({ + const auditorEvent = await auditor.createEvent({ eventId: 'entity-fetch', request: req, meta: { @@ -456,7 +453,7 @@ export async function createRouter( }, ) .post('/entities/by-refs', async (req, res) => { - const auditorEvent = await auditor?.createEvent({ + const auditorEvent = await auditor.createEvent({ eventId: 'entity-fetch', request: req, meta: { @@ -495,7 +492,7 @@ export async function createRouter( } }) .get('/entity-facets', async (req, res) => { - const auditorEvent = await auditor?.createEvent({ + const auditorEvent = await auditor.createEvent({ eventId: 'entity-facets', request: req, }); @@ -525,7 +522,7 @@ export async function createRouter( const location = await validateRequestBody(req, locationInput); const dryRun = yn(req.query.dryRun, { default: false }); - const auditorEvent = await auditor?.createEvent({ + const auditorEvent = await auditor.createEvent({ eventId: 'location-mutate', severityLevel: dryRun ? 'low' : 'medium', request: req, @@ -570,7 +567,7 @@ export async function createRouter( } }) .get('/locations', async (req, res) => { - const auditorEvent = await auditor?.createEvent({ + const auditorEvent = await auditor.createEvent({ eventId: 'location-fetch', request: req, meta: { @@ -597,7 +594,7 @@ export async function createRouter( .get('/locations/:id', async (req, res) => { const { id } = req.params; - const auditorEvent = await auditor?.createEvent({ + const auditorEvent = await auditor.createEvent({ eventId: 'location-fetch', request: req, meta: { @@ -628,7 +625,7 @@ export async function createRouter( .delete('/locations/:id', async (req, res) => { const { id } = req.params; - const auditorEvent = await auditor?.createEvent({ + const auditorEvent = await auditor.createEvent({ eventId: 'location-mutate', severityLevel: 'medium', request: req, @@ -659,7 +656,7 @@ export async function createRouter( const { kind, namespace, name } = req.params; const locationRef = `${kind}:${namespace}/${name}`; - const auditorEvent = await auditor?.createEvent({ + const auditorEvent = await auditor.createEvent({ eventId: 'location-fetch', request: req, meta: { @@ -692,7 +689,7 @@ export async function createRouter( if (locationAnalyzer) { router.post('/analyze-location', async (req, res) => { - const auditorEvent = await auditor?.createEvent({ + const auditorEvent = await auditor.createEvent({ eventId: 'location-analyze', request: req, }); @@ -743,86 +740,84 @@ export async function createRouter( }); } - if (orchestrator) { - router.post('/validate-entity', async (req, res) => { - const auditorEvent = await auditor?.createEvent({ - eventId: 'entity-validate', - request: req, + router.post('/validate-entity', async (req, res) => { + const auditorEvent = await auditor.createEvent({ + eventId: 'entity-validate', + request: req, + }); + + try { + const bodySchema = z.object({ + entity: z.unknown(), + location: z.string(), }); + let body: z.infer; + let entity: Entity; + let location: { type: string; target: string }; try { - const bodySchema = z.object({ - entity: z.unknown(), - location: z.string(), - }); - - let body: z.infer; - let entity: Entity; - let location: { type: string; target: string }; - try { - body = await validateRequestBody(req, bodySchema); - entity = validateEntityEnvelope(body.entity); - location = parseLocationRef(body.location); - if (location.type !== 'url') - throw new TypeError( - `Invalid location ref ${body.location}, only 'url:' is supported, e.g. url:https://host/path`, - ); - } catch (err) { - await auditorEvent?.fail({ - error: err, - }); - - return res.status(400).json({ - errors: [serializeError(err)], - }); - } - - const credentials = await httpAuth.credentials(req); - const authorizedValidationService = new AuthorizedValidationService( - orchestrator, - permissionsService, - ); - const processingResult = await authorizedValidationService.process( - { - entity: { - ...entity, - metadata: { - ...entity.metadata, - annotations: { - [ANNOTATION_LOCATION]: body.location, - [ANNOTATION_ORIGIN_LOCATION]: body.location, - ...entity.metadata.annotations, - }, - }, - }, - }, - credentials, - ); - - if (!processingResult.ok) { - const errors = processingResult.errors.map(e => serializeError(e)); - - await auditorEvent?.fail({ - // TODO(Rugvip): Seems like there aren't proper types for AggregateError yet - error: (AggregateError as any)(errors, 'Could not validate entity'), - }); - - res.status(400).json({ - errors, - }); - } - - await auditorEvent?.success(); - - return res.status(200).end(); + body = await validateRequestBody(req, bodySchema); + entity = validateEntityEnvelope(body.entity); + location = parseLocationRef(body.location); + if (location.type !== 'url') + throw new TypeError( + `Invalid location ref ${body.location}, only 'url:' is supported, e.g. url:https://host/path`, + ); } catch (err) { await auditorEvent?.fail({ error: err, }); - throw err; + + return res.status(400).json({ + errors: [serializeError(err)], + }); } - }); - } + + const credentials = await httpAuth.credentials(req); + const authorizedValidationService = new AuthorizedValidationService( + orchestrator, + permissionsService, + ); + const processingResult = await authorizedValidationService.process( + { + entity: { + ...entity, + metadata: { + ...entity.metadata, + annotations: { + [ANNOTATION_LOCATION]: body.location, + [ANNOTATION_ORIGIN_LOCATION]: body.location, + ...entity.metadata.annotations, + }, + }, + }, + }, + credentials, + ); + + if (!processingResult.ok) { + const errors = processingResult.errors.map(e => serializeError(e)); + + await auditorEvent?.fail({ + // TODO(Rugvip): Seems like there aren't proper types for AggregateError yet + error: (AggregateError as any)(errors, 'Could not validate entity'), + }); + + res.status(400).json({ + errors, + }); + } + + await auditorEvent?.success(); + + return res.status(200).end(); + } catch (err) { + await auditorEvent?.fail({ + error: err, + }); + throw err; + } + }); return router; } diff --git a/plugins/catalog-backend/src/tests/integration.test.ts b/plugins/catalog-backend/src/tests/integration.test.ts index 719c5e6ccf..486d1bfb99 100644 --- a/plugins/catalog-backend/src/tests/integration.test.ts +++ b/plugins/catalog-backend/src/tests/integration.test.ts @@ -296,6 +296,7 @@ class TestHarness { knex: options.db, orchestrator, stitcher, + scheduler: mockServices.scheduler(), createHash: () => createHash('sha1'), pollingIntervalMs: 50, onProcessingError: event => { From 498332ae5348bcabdbb63f0e4dccf565466e82fa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Mon, 22 Sep 2025 16:45:41 +0200 Subject: [PATCH 2/2] remove more unused things MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- .../DefaultProcessingDatabase.test.ts | 1 + .../src/database/DefaultProcessingDatabase.ts | 12 ++--- .../DefaultCatalogProcessingEngine.test.ts | 9 ++++ .../DefaultCatalogProcessingEngine.ts | 10 ++-- ...faultCatalogProcessingOrchestrator.test.ts | 9 +--- .../DefaultCatalogProcessingOrchestrator.ts | 4 -- .../src/service/CatalogBuilder.ts | 53 ++----------------- .../src/service/CatalogPlugin.ts | 3 +- .../src/service/DefaultRefreshService.test.ts | 2 + .../src/tests/integration.test.ts | 3 +- .../getProcessableEntitiesPerformance.test.ts | 1 + 11 files changed, 32 insertions(+), 75 deletions(-) diff --git a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.test.ts index a09e938575..790762a428 100644 --- a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.test.ts +++ b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.test.ts @@ -58,6 +58,7 @@ describe('DefaultProcessingDatabase', () => { minSeconds: 100, maxSeconds: 150, }), + events: mockServices.events.mock(), }), }; } diff --git a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts index 81c26c66ab..50c3cbb1c8 100644 --- a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts @@ -42,11 +42,7 @@ import { checkLocationKeyConflict } from './operations/refreshState/checkLocatio import { insertUnprocessedEntity } from './operations/refreshState/insertUnprocessedEntity'; import { updateUnprocessedEntity } from './operations/refreshState/updateUnprocessedEntity'; import { generateStableHash, generateTargetKey } from './util'; -import { - EventBroker, - EventParams, - EventsService, -} from '@backstage/plugin-events-node'; +import { EventParams, EventsService } from '@backstage/plugin-events-node'; import { DateTime } from 'luxon'; import { CATALOG_CONFLICTS_TOPIC } from '../constants'; import { CatalogConflictEventPayload } from '../catalog/types'; @@ -64,7 +60,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { database: Knex; logger: LoggerService; refreshInterval: ProcessingIntervalFunction; - eventBroker?: EventBroker | EventsService; + events: EventsService; }, ) { initDatabaseMetrics(options.database); @@ -367,7 +363,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { this.options.logger.warn( `Detected conflicting entityRef ${entityRef} already referenced by ${conflictingKey} and now also ${locationKey}`, ); - if (this.options.eventBroker && locationKey) { + if (locationKey) { const eventParams: EventParams = { topic: CATALOG_CONFLICTS_TOPIC, eventPayload: { @@ -378,7 +374,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { lastConflictAt: DateTime.now().toISO()!, }, }; - await this.options.eventBroker?.publish(eventParams); + await this.options.events.publish(eventParams); } } } diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts index 93c22aab1e..5e6312ec4a 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts @@ -71,6 +71,7 @@ describe('DefaultCatalogProcessingEngine', () => { stitcher: stitcher, createHash: () => hash, scheduler: mockServices.scheduler(), + events: mockServices.events.mock(), }); db.transaction.mockImplementation(cb => cb((() => {}) as any)); @@ -139,6 +140,7 @@ describe('DefaultCatalogProcessingEngine', () => { stitcher: stitcher, scheduler: mockServices.scheduler(), createHash: () => hash, + events: mockServices.events.mock(), }); db.transaction.mockImplementation(cb => cb((() => {}) as any)); @@ -223,6 +225,7 @@ describe('DefaultCatalogProcessingEngine', () => { stitcher: stitcher, scheduler: mockServices.scheduler(), createHash: () => hash, + events: mockServices.events.mock(), }); db.transaction.mockImplementation(cb => cb((() => {}) as any)); @@ -301,6 +304,7 @@ describe('DefaultCatalogProcessingEngine', () => { stitcher: stitcher, scheduler: mockServices.scheduler(), createHash: () => hash, + events: mockServices.events.mock(), }); db.transaction.mockImplementation(cb => cb((() => {}) as any)); @@ -362,6 +366,7 @@ describe('DefaultCatalogProcessingEngine', () => { scheduler: mockServices.scheduler(), createHash: () => hash, pollingIntervalMs: 100, + events: mockServices.events.mock(), }); db.transaction.mockImplementation(cb => cb((() => {}) as any)); @@ -478,6 +483,7 @@ describe('DefaultCatalogProcessingEngine', () => { scheduler: mockServices.scheduler(), createHash: () => hash, pollingIntervalMs: 100, + events: mockServices.events.mock(), }); db.transaction.mockImplementation(cb => cb((() => {}) as any)); @@ -584,6 +590,7 @@ describe('DefaultCatalogProcessingEngine', () => { scheduler: mockServices.scheduler(), createHash: () => hash, pollingIntervalMs: 100, + events: mockServices.events.mock(), }); db.transaction.mockImplementation(cb => cb((() => {}) as any)); @@ -668,6 +675,7 @@ describe('DefaultCatalogProcessingEngine', () => { scheduler: mockServices.scheduler(), createHash: () => hash, pollingIntervalMs: 100, + events: mockServices.events.mock(), }); db.transaction.mockImplementation(cb => cb((() => {}) as any)); @@ -757,6 +765,7 @@ describe('DefaultCatalogProcessingEngine', () => { scheduler: mockServices.scheduler(), createHash: () => hash, pollingIntervalMs: 100, + events: mockServices.events.mock(), }); db.transaction.mockImplementation(cb => cb((() => {}) as any)); diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts index 7061309175..b2a562ea6a 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts @@ -36,7 +36,7 @@ import { withActiveSpan, } from '../util/opentelemetry'; import { deleteOrphanedEntities } from '../database/operations/util/deleteOrphanedEntities'; -import { EventBroker, EventsService } from '@backstage/plugin-events-node'; +import { EventsService } from '@backstage/plugin-events-node'; import { CATALOG_ERRORS_TOPIC } from '../constants'; import { LoggerService, SchedulerService } from '@backstage/backend-plugin-api'; @@ -73,7 +73,7 @@ export class DefaultCatalogProcessingEngine { errors: Error[]; }) => Promise | void; private readonly tracker: ProgressTracker; - private readonly eventBroker?: EventBroker | EventsService; + private readonly events: EventsService; private stopFunc?: () => void; @@ -93,7 +93,7 @@ export class DefaultCatalogProcessingEngine { errors: Error[]; }) => Promise | void; tracker?: ProgressTracker; - eventBroker?: EventBroker | EventsService; + events: EventsService; }) { this.config = options.config; this.scheduler = options.scheduler; @@ -107,7 +107,7 @@ export class DefaultCatalogProcessingEngine { this.orphanCleanupIntervalMs = options.orphanCleanupIntervalMs ?? 30_000; this.onProcessingError = options.onProcessingError; this.tracker = options.tracker ?? progressTracker(); - this.eventBroker = options.eventBroker; + this.events = options.events; this.stopFunc = undefined; } @@ -201,7 +201,7 @@ export class DefaultCatalogProcessingEngine { const location = unprocessedEntity?.metadata?.annotations?.[ANNOTATION_LOCATION]; if (result.errors.length) { - this.eventBroker?.publish({ + this.events.publish({ topic: CATALOG_ERRORS_TOPIC, eventPayload: { entity: entityRef, diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.test.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.test.ts index 31d7c34b94..beb3611206 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.test.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.test.ts @@ -97,7 +97,6 @@ describe('DefaultCatalogProcessingOrchestrator', () => { parser: defaultEntityDataParser, policy: EntityPolicies.allOf([]), rulesEnforcer: { isAllowed: () => true }, - legacySingleProcessorValidation: false, }); it('runs a minimal processing', async () => { @@ -192,7 +191,7 @@ describe('DefaultCatalogProcessingOrchestrator', () => { }); }); - it('runs all processor validations when asked to', async () => { + it('runs all processor validations', async () => { const validate = jest.fn(async () => true); const processor1: CatalogProcessor = { getProcessorName: () => 'processor1', @@ -213,7 +212,6 @@ describe('DefaultCatalogProcessingOrchestrator', () => { parser: defaultEntityDataParser, policy: EntityPolicies.allOf([]), rulesEnforcer: { isAllowed: () => true }, - legacySingleProcessorValidation: true, }); const modern = new DefaultCatalogProcessingOrchestrator({ @@ -226,13 +224,12 @@ describe('DefaultCatalogProcessingOrchestrator', () => { parser: defaultEntityDataParser, policy: EntityPolicies.allOf([]), rulesEnforcer: { isAllowed: () => true }, - legacySingleProcessorValidation: false, }); await expect(legacy.process({ entity })).resolves.toMatchObject({ ok: true, }); - expect(validate).toHaveBeenCalledTimes(1); + expect(validate).toHaveBeenCalledTimes(2); validate.mockClear(); @@ -291,7 +288,6 @@ describe('DefaultCatalogProcessingOrchestrator', () => { parser, policy: EntityPolicies.allOf([]), rulesEnforcer, - legacySingleProcessorValidation: false, }); rulesEnforcer.isAllowed.mockReturnValueOnce(true); @@ -333,7 +329,6 @@ describe('DefaultCatalogProcessingOrchestrator', () => { parser, policy: EntityPolicies.allOf([new FailingEntityPolicy()]), rulesEnforcer, - legacySingleProcessorValidation: false, }); await expect( diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts index b1fcba5d40..79873264a3 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts @@ -95,7 +95,6 @@ export class DefaultCatalogProcessingOrchestrator parser: CatalogProcessorParser; policy: EntityPolicy; rulesEnforcer: CatalogRulesEnforcer; - legacySingleProcessorValidation: boolean; }, ) {} @@ -309,9 +308,6 @@ export class DefaultCatalogProcessingOrchestrator ); if (thisValid) { valid = true; - if (this.options.legacySingleProcessorValidation) { - break; - } } } catch (e) { throw new InputError( diff --git a/plugins/catalog-backend/src/service/CatalogBuilder.ts b/plugins/catalog-backend/src/service/CatalogBuilder.ts index 40fcc3a624..0b35e84102 100644 --- a/plugins/catalog-backend/src/service/CatalogBuilder.ts +++ b/plugins/catalog-backend/src/service/CatalogBuilder.ts @@ -55,7 +55,7 @@ import { PlaceholderResolver, ScmLocationAnalyzer, } from '@backstage/plugin-catalog-node'; -import { EventBroker, EventsService } from '@backstage/plugin-events-node'; +import { EventsService } from '@backstage/plugin-events-node'; import { Permission, PermissionAuthorizer, @@ -123,6 +123,7 @@ export type CatalogEnvironment = { auth: AuthService; httpAuth: HttpAuthService; auditor: AuditorService; + events: EventsService; }; /** @@ -168,8 +169,6 @@ export class CatalogBuilder { private readonly permissions: Permission[]; private readonly permissionRules: CatalogPermissionRuleInput[]; private allowedLocationType: string[]; - private legacySingleProcessorValidation = false; - private eventBroker?: EventBroker | EventsService; /** * Creates a catalog builder. @@ -216,31 +215,6 @@ export class CatalogBuilder { return this; } - /** - * Processing interval determines how often entities should be processed. - * Seconds provided will be multiplied by 1.5 - * The default processing interval is 100-150 seconds. - * setting this too low will potentially deplete request quotas to upstream services. - */ - setProcessingIntervalSeconds(seconds: number): CatalogBuilder { - this.processingInterval = createRandomProcessingInterval({ - minSeconds: seconds, - maxSeconds: seconds * 1.5, - }); - return this; - } - - /** - * Overwrites the default processing interval function used to spread - * entity updates in the catalog. - */ - setProcessingInterval( - processingInterval: ProcessingIntervalFunction, - ): CatalogBuilder { - this.processingInterval = processingInterval; - return this; - } - /** * Overwrites the default location analyzer. */ @@ -427,23 +401,6 @@ export class CatalogBuilder { return this; } - /** - * Enables the legacy behaviour of canceling validation early whenever only a - * single processor declares an entity kind to be valid. - */ - useLegacySingleProcessorValidation(): this { - this.legacySingleProcessorValidation = true; - return this; - } - - /** - * Enables the publishing of events for conflicts in the DefaultProcessingDatabase - */ - setEventBroker(broker: EventBroker | EventsService): CatalogBuilder { - this.eventBroker = broker; - return this; - } - /** * Wires up and returns all of the component parts of the catalog */ @@ -461,6 +418,7 @@ export class CatalogBuilder { auditor, auth, httpAuth, + events, } = this.env; const enableRelationsCompatibility = Boolean( @@ -485,8 +443,8 @@ export class CatalogBuilder { const processingDatabase = new DefaultProcessingDatabase({ database: dbClient, logger, + events, refreshInterval: this.processingInterval, - eventBroker: this.eventBroker, }); const providerDatabase = new DefaultProviderDatabase({ database: dbClient, @@ -523,7 +481,6 @@ export class CatalogBuilder { logger, parser, policy, - legacySingleProcessorValidation: this.legacySingleProcessorValidation, }); const entitiesCatalog = new AuthorizedEntitiesCatalog( @@ -588,7 +545,7 @@ export class CatalogBuilder { onProcessingError: event => { this.onProcessingError?.(event); }, - eventBroker: this.eventBroker, + events, }); const locationAnalyzer = diff --git a/plugins/catalog-backend/src/service/CatalogPlugin.ts b/plugins/catalog-backend/src/service/CatalogPlugin.ts index a9bb2c9285..24147e3cc5 100644 --- a/plugins/catalog-backend/src/service/CatalogPlugin.ts +++ b/plugins/catalog-backend/src/service/CatalogPlugin.ts @@ -271,10 +271,9 @@ export const catalogPlugin = createBackendPlugin({ auth, httpAuth, auditor, + events, }); - builder.setEventBroker(events); - if (processingExtensions.onProcessingErrorHandler) { builder.subscribe({ onProcessingError: processingExtensions.onProcessingErrorHandler, diff --git a/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts b/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts index f999400b8a..29519e9e6b 100644 --- a/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts +++ b/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts @@ -57,6 +57,7 @@ describe('DefaultRefreshService', () => { database: knex, logger, refreshInterval: () => 100, + events: mockServices.events.mock(), }), catalogDb: new DefaultCatalogDatabase({ database: knex, @@ -162,6 +163,7 @@ describe('DefaultRefreshService', () => { }, createHash: () => createHash('sha1'), pollingIntervalMs: 50, + events: mockServices.events.mock(), }); return engine; diff --git a/plugins/catalog-backend/src/tests/integration.test.ts b/plugins/catalog-backend/src/tests/integration.test.ts index 486d1bfb99..39871583af 100644 --- a/plugins/catalog-backend/src/tests/integration.test.ts +++ b/plugins/catalog-backend/src/tests/integration.test.ts @@ -244,6 +244,7 @@ class TestHarness { const processingDatabase = new DefaultProcessingDatabase({ database: options.db, logger, + events: mockServices.events.mock(), refreshInterval: () => 0.05, }); @@ -273,7 +274,6 @@ class TestHarness { logger, parser: defaultEntityDataParser, policy: EntityPolicies.allOf([]), - legacySingleProcessorValidation: false, }); const stitcher = DefaultStitcher.fromConfig(config, { knex: options.db, @@ -303,6 +303,7 @@ class TestHarness { proxyProgressTracker.reportError(event.unprocessedEntity, event.errors); }, tracker: proxyProgressTracker, + events: mockServices.events.mock(), }); const refresh = new DefaultRefreshService({ database: catalogDatabase }); diff --git a/plugins/catalog-backend/src/tests/performance/getProcessableEntitiesPerformance.test.ts b/plugins/catalog-backend/src/tests/performance/getProcessableEntitiesPerformance.test.ts index e2ed160b42..a1cc1d3238 100644 --- a/plugins/catalog-backend/src/tests/performance/getProcessableEntitiesPerformance.test.ts +++ b/plugins/catalog-backend/src/tests/performance/getProcessableEntitiesPerformance.test.ts @@ -79,6 +79,7 @@ describePerformanceTest('getProcessableEntities', () => { const sut = new DefaultProcessingDatabase({ database: knex, logger: mockServices.logger.mock(), + events: mockServices.events.mock(), refreshInterval: () => 0, });