diff --git a/.changeset/odd-falcons-shout.md b/.changeset/odd-falcons-shout.md new file mode 100644 index 0000000000..bfbd578c60 --- /dev/null +++ b/.changeset/odd-falcons-shout.md @@ -0,0 +1,38 @@ +--- +'@backstage/plugin-catalog-backend': minor +--- + +The catalog now has a new, optional `catalog.orphanStrategy` app-config parameter, which can have the string values `'keep'` (default) or `'delete'`. + +If set to `'keep'` or left unset, the old behavior is maintained of keeping orphaned entities around until manually deleted. + +If set to `'delete'`, the catalog will attempt to automatically clean out orphaned entities without manual intervention. Note that there are no guarantees that this process is instantaneous, so there may be some delay before orphaned items disappear. + +For context, the [Life of an Entity](https://backstage.io/docs/features/software-catalog/life-of-an-entity/#orphaning) article goes into some more details on how the nature of orphaning works. + +To enable the new behavior, you will need to pass the plugin task scheduler to your catalog backend builder. If your code already looks like this, you don't need to change it: + +```ts +// in packages/backend/src/plugins/catalog.ts +export default async function createPlugin( + env: PluginEnvironment, +): Promise { + const builder = await CatalogBuilder.create(env); +``` + +But if you pass things into the catalog builder one by one, you'll need to add the new field: + +```diff + // in packages/backend/src/plugins/catalog.ts + const builder = await CatalogBuilder.create({ + // ... other dependencies ++ scheduler: env.scheduler, + }); +``` + +Finally adjust your app-config: + +```yaml +catalog: + orphanStrategy: delete +``` diff --git a/plugins/catalog-backend/api-report.md b/plugins/catalog-backend/api-report.md index d491f391d5..e3ef784a38 100644 --- a/plugins/catalog-backend/api-report.md +++ b/plugins/catalog-backend/api-report.md @@ -46,6 +46,7 @@ import { PermissionRule } from '@backstage/plugin-permission-node'; import { PermissionRuleParams } from '@backstage/plugin-permission-common'; import { PluginDatabaseManager } from '@backstage/backend-common'; import { PluginEndpointDiscovery } from '@backstage/backend-common'; +import { PluginTaskScheduler } from '@backstage/backend-tasks'; import { Router } from 'express'; import { ScmIntegrationRegistry } from '@backstage/integration'; import { TokenManager } from '@backstage/backend-common'; @@ -172,6 +173,7 @@ export type CatalogEnvironment = { config: Config; reader: UrlReader; permissions: PermissionEvaluator | PermissionAuthorizer; + scheduler?: PluginTaskScheduler; }; // @public diff --git a/plugins/catalog-backend/config.d.ts b/plugins/catalog-backend/config.d.ts index 5c5bfea753..03e47f8e61 100644 --- a/plugins/catalog-backend/config.d.ts +++ b/plugins/catalog-backend/config.d.ts @@ -81,7 +81,6 @@ export interface Config { * be used in combination with static locations to only serve operator * provided locations. Effectively this removes the ability to register new * components to a running backstage instance. - * */ readonly?: boolean; @@ -136,5 +135,12 @@ export interface Config { allow: Array; }>; }>; + + /** + * The strategy to use for entities that are orphaned, i.e. no longer have + * any other entities or providers referencing them. The default value is + * "keep". + */ + orphanStrategy?: 'keep' | 'delete'; }; } diff --git a/plugins/catalog-backend/package.json b/plugins/catalog-backend/package.json index d8b42fafe6..b155051d74 100644 --- a/plugins/catalog-backend/package.json +++ b/plugins/catalog-backend/package.json @@ -47,6 +47,7 @@ "dependencies": { "@backstage/backend-common": "workspace:^", "@backstage/backend-plugin-api": "workspace:^", + "@backstage/backend-tasks": "workspace:^", "@backstage/catalog-client": "workspace:^", "@backstage/catalog-model": "workspace:^", "@backstage/config": "workspace:^", diff --git a/plugins/catalog-backend/src/alpha.ts b/plugins/catalog-backend/src/alpha.ts index aa1309aecd..c90f4d1cdf 100644 --- a/plugins/catalog-backend/src/alpha.ts +++ b/plugins/catalog-backend/src/alpha.ts @@ -15,7 +15,7 @@ */ // TODO(Rugvip): Re-exported for alpha types as the API report will otherwise -// produce warnings due to the indirect dependency. We be nice to avoid. +// produce warnings due to the indirect dependency. Would be nice to avoid. import type { EntitiesSearchFilter } from './catalog/types'; export type { /** @alpha */ EntitiesSearchFilter }; diff --git a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts index 761f5bfd72..398efa5b4b 100644 --- a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts @@ -16,6 +16,7 @@ import { Entity, stringifyEntityRef } from '@backstage/catalog-model'; import { ConflictError } from '@backstage/errors'; +import { DeferredEntity } from '@backstage/plugin-catalog-node'; import { Knex } from 'knex'; import lodash from 'lodash'; import type { Logger } from 'winston'; @@ -38,11 +39,10 @@ import { UpdateEntityCacheOptions, UpdateProcessedEntityOptions, } from './types'; - -import { DeferredEntity } from '@backstage/plugin-catalog-node'; import { checkLocationKeyConflict } from './operations/refreshState/checkLocationKeyConflict'; import { insertUnprocessedEntity } from './operations/refreshState/insertUnprocessedEntity'; import { updateUnprocessedEntity } from './operations/refreshState/updateUnprocessedEntity'; +import { deleteOrphanedEntities } from './operations/util/deleteOrphanedEntities'; import { generateStableHash } from './util'; // The number of items that are sent per batch to the database layer, when @@ -269,6 +269,11 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { return { entityRefs }; } + async deleteOrphanedEntities(txOpaque: Transaction): Promise { + const tx = txOpaque as Knex.Transaction; + return await deleteOrphanedEntities({ tx }); + } + async transaction(fn: (tx: Transaction) => Promise): Promise { try { let result: T | undefined = undefined; diff --git a/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts b/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts new file mode 100644 index 0000000000..243b5db1b9 --- /dev/null +++ b/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.test.ts @@ -0,0 +1,196 @@ +/* + * Copyright 2023 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 { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils'; +import { Knex } from 'knex'; +import * as uuid from 'uuid'; +import { applyDatabaseMigrations } from '../../migrations'; +import { + DbFinalEntitiesRow, + DbRefreshStateReferencesRow, + DbRefreshStateRow, + DbRelationsRow, +} from '../../tables'; +import { deleteOrphanedEntities } from './deleteOrphanedEntities'; + +jest.setTimeout(60_000); + +describe('deleteOrphanedEntities', () => { + const databases = TestDatabases.create({ + ids: ['MYSQL_8', 'POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'], + }); + + async function createDatabase(databaseId: TestDatabaseId) { + const knex = await databases.init(databaseId); + await applyDatabaseMigrations(knex); + return knex; + } + + async function run(knex: Knex): Promise { + let result: number; + await knex.transaction( + async tx => { + // We can't return here, as knex swallows the return type in case the + // transaction is rolled back: + // https://github.com/knex/knex/blob/e37aeaa31c8ef9c1b07d2e4d3ec6607e557d800d/lib/transaction.js#L136 + result = await deleteOrphanedEntities({ tx }); + }, + { + // If we explicitly trigger a rollback, don't fail. + doNotRejectOnRollback: true, + }, + ); + return result!; + } + + async function insertEntity(knex: Knex, ...entityRefs: string[]) { + for (const ref of entityRefs) { + const entityId = uuid.v4(); + await knex('refresh_state').insert({ + entity_id: entityId, + entity_ref: ref, + unprocessed_entity: '{}', + processed_entity: '{}', + errors: '[]', + next_update_at: '2021-04-01 13:37:00', + last_discovery_at: '2021-04-01 13:37:00', + result_hash: 'original', + }); + await knex('final_entities').insert({ + entity_id: entityId, + hash: 'original', + stitch_ticket: '', + }); + } + } + + async function insertReference( + knex: Knex, + ...refs: DbRefreshStateReferencesRow[] + ) { + await knex('refresh_state_references').insert( + refs, + ); + } + + async function insertRelation(knex: Knex, fromRef: string, toRef: string) { + const orig = await knex + .select('entity_id') + .from('refresh_state') + .where('entity_ref', fromRef); + await knex('relations').insert({ + originating_entity_id: orig[0].entity_id, + type: 'fake', + source_entity_ref: fromRef, + target_entity_ref: toRef, + }); + } + + async function refreshState(knex: Knex) { + return await knex('refresh_state') + .orderBy('entity_ref') + .select('entity_ref', 'result_hash'); + } + + async function finalEntities(knex: Knex) { + return await knex('final_entities') + .join( + 'refresh_state', + 'final_entities.entity_id', + 'refresh_state.entity_id', + ) + .orderBy('refresh_state.entity_ref') + .select({ + entity_ref: 'refresh_state.entity_ref', + hash: 'final_entities.hash', + }); + } + + it.each(databases.eachSupportedId())( + 'works for some mixed paths, %p', + async databaseId => { + /* + In this graph, edges represent refresh state references, not entity relations: + + P1 - E1 -- E2 + / + E3 + / + E4 + \ + E5 + / + E6 + \ + E7 + / + P2 - E8 + + P3 - E9 + + E10 + + Result: E3, E4, E5, E6, and E10 deleted; others remain + Entities that had relations pointing at orphans are marked for reprocessing + */ + const knex = await createDatabase(databaseId); + await insertEntity( + knex, + 'E1', + 'E2', + 'E3', + 'E4', + 'E5', + 'E6', + 'E7', + 'E8', + 'E9', + 'E10', + ); + await insertReference( + knex, + { source_key: 'P1', target_entity_ref: 'E1' }, + { source_entity_ref: 'E1', target_entity_ref: 'E2' }, + { source_entity_ref: 'E3', target_entity_ref: 'E2' }, + { source_entity_ref: 'E4', target_entity_ref: 'E3' }, + { source_entity_ref: 'E4', target_entity_ref: 'E5' }, + { source_entity_ref: 'E6', target_entity_ref: 'E5' }, + { source_entity_ref: 'E6', target_entity_ref: 'E7' }, + { source_key: 'P2', target_entity_ref: 'E8' }, + { source_entity_ref: 'E8', target_entity_ref: 'E7' }, + { source_key: 'P3', target_entity_ref: 'E9' }, + ); + await insertRelation(knex, 'E1', 'E2'); + await insertRelation(knex, 'E2', 'E3'); + await insertRelation(knex, 'E10', 'E6'); + await expect(run(knex)).resolves.toEqual(5); + await expect(refreshState(knex)).resolves.toEqual([ + { entity_ref: 'E1', result_hash: 'original' }, + { entity_ref: 'E2', result_hash: 'orphan-relation-deleted' }, + { entity_ref: 'E7', result_hash: 'original' }, + { entity_ref: 'E8', result_hash: 'original' }, + { entity_ref: 'E9', result_hash: 'original' }, + ]); + await expect(finalEntities(knex)).resolves.toEqual([ + { entity_ref: 'E1', hash: 'original' }, + { entity_ref: 'E2', hash: 'orphan-relation-deleted' }, + { entity_ref: 'E7', hash: 'original' }, + { entity_ref: 'E8', hash: 'original' }, + { entity_ref: 'E9', hash: 'original' }, + ]); + }, + ); +}); diff --git a/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.ts b/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.ts new file mode 100644 index 0000000000..920acae7bd --- /dev/null +++ b/plugins/catalog-backend/src/database/operations/util/deleteOrphanedEntities.ts @@ -0,0 +1,95 @@ +/* + * Copyright 2023 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 { Knex } from 'knex'; +import uniq from 'lodash/uniq'; +import { DbFinalEntitiesRow, DbRefreshStateRow } from '../../tables'; + +/** + * Finds and deletes all orphaned entities, i.e. entities that do not have any + * incoming references to them, and also eagerly deletes all of their children + * that would otherwise become orphaned. + */ +export async function deleteOrphanedEntities(options: { + tx: Knex.Transaction | Knex; +}): Promise { + const { tx } = options; + + let total = 0; + + // Limit iterations for sanity + for (let i = 0; i < 100; ++i) { + const candidates = await tx + .with('orphans', orphans => + orphans + .from('refresh_state') + .select('entity_id', 'entity_ref') + .whereNotIn('entity_ref', keep => + keep.distinct('target_entity_ref').from('refresh_state_references'), + ), + ) + .select({ + entityId: 'orphans.entity_id', + relationSourceId: 'refresh_state.entity_id', + }) + .from('orphans') + .leftOuterJoin( + 'relations', + 'relations.target_entity_ref', + 'orphans.entity_ref', + ) + .leftOuterJoin( + 'refresh_state', + 'refresh_state.entity_ref', + 'relations.source_entity_ref', + ); + + if (!candidates.length) { + break; + } + + const orphanIds: string[] = uniq(candidates.map(r => r.entityId)); + const orphanRelationIds: string[] = uniq( + candidates.map(r => r.relationSourceId).filter(Boolean), + ); + + total += orphanIds.length; + + // Delete the orphans themselves + await tx + .table('refresh_state') + .delete() + .whereIn('entity_id', orphanIds); + + // Mark all of things that the orphans had relations to for processing and + // stitching + await tx + .table('final_entities') + .update({ + hash: 'orphan-relation-deleted', + }) + .whereIn('entity_id', orphanRelationIds); + await tx + .table('refresh_state') + .update({ + result_hash: 'orphan-relation-deleted', + next_update_at: tx.fn.now(), + }) + .whereIn('entity_id', orphanRelationIds); + } + + return total; +} diff --git a/plugins/catalog-backend/src/database/types.ts b/plugins/catalog-backend/src/database/types.ts index 56e5f48f14..592d5fe063 100644 --- a/plugins/catalog-backend/src/database/types.ts +++ b/plugins/catalog-backend/src/database/types.ts @@ -151,6 +151,8 @@ export interface ProcessingDatabase { txOpaque: Transaction, options: ListParentsOptions, ): Promise; + + deleteOrphanedEntities(txOpaque: Transaction): Promise; } /** diff --git a/plugins/catalog-backend/src/integration.test.ts b/plugins/catalog-backend/src/integration.test.ts index b13e6ef9de..9aae415829 100644 --- a/plugins/catalog-backend/src/integration.test.ts +++ b/plugins/catalog-backend/src/integration.test.ts @@ -277,18 +277,19 @@ class TestHarness { new NoopProgressTracker(), ); - const engine = new DefaultCatalogProcessingEngine( + const engine = new DefaultCatalogProcessingEngine({ + config: new ConfigReader({}), logger, processingDatabase, orchestrator, stitcher, - () => createHash('sha1'), - 50, - event => { + createHash: () => createHash('sha1'), + pollingIntervalMs: 50, + onProcessingError: event => { proxyProgressTracker.reportError(event.unprocessedEntity, event.errors); }, - proxyProgressTracker, - ); + tracker: proxyProgressTracker, + }); const refresh = new DefaultRefreshService({ database: catalogDatabase }); diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts index 3002218d89..00235ef601 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.test.ts @@ -22,6 +22,7 @@ import { DefaultProcessingDatabase } from '../database/DefaultProcessingDatabase import { DefaultCatalogProcessingEngine } from './DefaultCatalogProcessingEngine'; import { CatalogProcessingOrchestrator } from './types'; import { Stitcher } from '../stitching/Stitcher'; +import { ConfigReader } from '@backstage/config'; describe('DefaultCatalogProcessingEngine', () => { const db = { @@ -61,13 +62,14 @@ describe('DefaultCatalogProcessingEngine', () => { state: {}, refreshKeys: [], }); - const engine = new DefaultCatalogProcessingEngine( - getVoidLogger(), - db, - orchestrator, - stitcher, - () => hash, - ); + const engine = new DefaultCatalogProcessingEngine({ + config: new ConfigReader({}), + logger: getVoidLogger(), + processingDatabase: db, + orchestrator: orchestrator, + stitcher: stitcher, + createHash: () => hash, + }); db.transaction.mockImplementation(cb => cb((() => {}) as any)); @@ -127,13 +129,14 @@ describe('DefaultCatalogProcessingEngine', () => { state: {}, refreshKeys: [], }); - const engine = new DefaultCatalogProcessingEngine( - getVoidLogger(), - db, - orchestrator, - stitcher, - () => hash, - ); + const engine = new DefaultCatalogProcessingEngine({ + config: new ConfigReader({}), + logger: getVoidLogger(), + processingDatabase: db, + orchestrator: orchestrator, + stitcher: stitcher, + createHash: () => hash, + }); db.transaction.mockImplementation(cb => cb((() => {}) as any)); @@ -209,13 +212,14 @@ describe('DefaultCatalogProcessingEngine', () => { refreshKeys: [], }); - const engine = new DefaultCatalogProcessingEngine( - getVoidLogger(), - db, - orchestrator, - stitcher, - () => hash, - ); + const engine = new DefaultCatalogProcessingEngine({ + config: new ConfigReader({}), + logger: getVoidLogger(), + processingDatabase: db, + orchestrator: orchestrator, + stitcher: stitcher, + createHash: () => hash, + }); db.transaction.mockImplementation(cb => cb((() => {}) as any)); @@ -284,13 +288,14 @@ describe('DefaultCatalogProcessingEngine', () => { errors: [], }); - const engine = new DefaultCatalogProcessingEngine( - getVoidLogger(), - db, - orchestrator, - stitcher, - () => hash, - ); + const engine = new DefaultCatalogProcessingEngine({ + config: new ConfigReader({}), + logger: getVoidLogger(), + processingDatabase: db, + orchestrator: orchestrator, + stitcher: stitcher, + createHash: () => hash, + }); db.transaction.mockImplementation(cb => cb((() => {}) as any)); @@ -341,14 +346,15 @@ describe('DefaultCatalogProcessingEngine', () => { }); it('should stitch both the previous and new sources when relations change', async () => { - const engine = new DefaultCatalogProcessingEngine( - getVoidLogger(), - db, - orchestrator, - stitcher, - () => hash, - 100, - ); + const engine = new DefaultCatalogProcessingEngine({ + config: new ConfigReader({}), + logger: getVoidLogger(), + processingDatabase: db, + orchestrator: orchestrator, + stitcher: stitcher, + createHash: () => hash, + pollingIntervalMs: 100, + }); db.transaction.mockImplementation(cb => cb((() => {}) as any)); @@ -454,14 +460,15 @@ describe('DefaultCatalogProcessingEngine', () => { }); it('should not stitch sources entities when relations are the same', async () => { - const engine = new DefaultCatalogProcessingEngine( - getVoidLogger(), - db, - orchestrator, - stitcher, - () => hash, - 100, - ); + const engine = new DefaultCatalogProcessingEngine({ + config: new ConfigReader({}), + logger: getVoidLogger(), + processingDatabase: db, + orchestrator: orchestrator, + stitcher: stitcher, + createHash: () => hash, + pollingIntervalMs: 100, + }); db.transaction.mockImplementation(cb => cb((() => {}) as any)); @@ -535,14 +542,15 @@ describe('DefaultCatalogProcessingEngine', () => { }); it('should stitch sources entities when new relation of different type added', async () => { - const engine = new DefaultCatalogProcessingEngine( - getVoidLogger(), - db, - orchestrator, - stitcher, - () => hash, - 100, - ); + const engine = new DefaultCatalogProcessingEngine({ + config: new ConfigReader({}), + logger: getVoidLogger(), + processingDatabase: db, + orchestrator: orchestrator, + stitcher: stitcher, + createHash: () => hash, + pollingIntervalMs: 100, + }); db.transaction.mockImplementation(cb => cb((() => {}) as any)); @@ -621,14 +629,15 @@ describe('DefaultCatalogProcessingEngine', () => { }); it('should stitch sources entities when relation is removed', async () => { - const engine = new DefaultCatalogProcessingEngine( - getVoidLogger(), - db, - orchestrator, - stitcher, - () => hash, - 100, - ); + const engine = new DefaultCatalogProcessingEngine({ + config: new ConfigReader({}), + logger: getVoidLogger(), + processingDatabase: db, + orchestrator: orchestrator, + stitcher: stitcher, + createHash: () => hash, + pollingIntervalMs: 100, + }); 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 8457381ccd..ab1e0b96c9 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts @@ -33,34 +33,85 @@ import { } from './types'; import { Stitcher } from '../stitching/Stitcher'; import { startTaskPipeline } from './TaskPipeline'; +import { PluginTaskScheduler } from '@backstage/backend-tasks'; +import { Config } from '@backstage/config'; const CACHE_TTL = 5; export type ProgressTracker = ReturnType; export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { + private readonly config: Config; + private readonly scheduler?: PluginTaskScheduler; + private readonly logger: Logger; + private readonly processingDatabase: ProcessingDatabase; + private readonly orchestrator: CatalogProcessingOrchestrator; + private readonly stitcher: Stitcher; + private readonly createHash: () => Hash; + private readonly pollingIntervalMs: number; + private readonly orphanCleanupIntervalMs: number; + private readonly onProcessingError?: (event: { + unprocessedEntity: Entity; + errors: Error[]; + }) => Promise | void; + private readonly tracker: ProgressTracker; + private stopFunc?: () => void; - constructor( - private readonly logger: Logger, - private readonly processingDatabase: ProcessingDatabase, - private readonly orchestrator: CatalogProcessingOrchestrator, - private readonly stitcher: Stitcher, - private readonly createHash: () => Hash, - private readonly pollingIntervalMs: number = 1000, - private readonly onProcessingError?: (event: { + constructor(options: { + config: Config; + scheduler?: PluginTaskScheduler; + logger: Logger; + processingDatabase: ProcessingDatabase; + orchestrator: CatalogProcessingOrchestrator; + stitcher: Stitcher; + createHash: () => Hash; + pollingIntervalMs?: number; + orphanCleanupIntervalMs?: number; + onProcessingError?: (event: { unprocessedEntity: Entity; errors: Error[]; - }) => Promise | void, - private readonly tracker: ProgressTracker = progressTracker(), - ) {} + }) => Promise | void; + tracker?: ProgressTracker; + }) { + this.config = options.config; + this.scheduler = options.scheduler; + this.logger = options.logger; + this.processingDatabase = options.processingDatabase; + this.orchestrator = options.orchestrator; + this.stitcher = options.stitcher; + this.createHash = options.createHash; + this.pollingIntervalMs = options.pollingIntervalMs ?? 1_000; + this.orphanCleanupIntervalMs = options.orphanCleanupIntervalMs ?? 30_000; + this.onProcessingError = options.onProcessingError; + this.tracker = options.tracker ?? progressTracker(); + + this.stopFunc = undefined; + } async start() { if (this.stopFunc) { throw new Error('Processing engine is already started'); } - this.stopFunc = startTaskPipeline({ + const stopPipeline = this.startPipeline(); + const stopCleanup = this.startOrphanCleanup(); + + this.stopFunc = () => { + stopPipeline(); + stopCleanup(); + }; + } + + async stop() { + if (this.stopFunc) { + this.stopFunc(); + this.stopFunc = undefined; + } + } + + private startPipeline(): () => void { + return startTaskPipeline({ lowWatermark: 5, highWatermark: 10, pollingIntervalMs: this.pollingIntervalMs, @@ -252,11 +303,44 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { }); } - async stop() { - if (this.stopFunc) { - this.stopFunc(); - this.stopFunc = undefined; + private startOrphanCleanup(): () => void { + const strategy = + this.config.getOptionalString('catalog.orphanStrategy') ?? 'keep'; + if (strategy !== 'delete') { + return () => {}; } + + const runOnce = async () => { + try { + await this.processingDatabase.transaction(async tx => { + const n = await this.processingDatabase.deleteOrphanedEntities(tx); + this.logger.info(`Deleted ${n} orphaned entities`); + }); + } catch (error) { + this.logger.warn(`Failed to delete orphaned entities`, error); + } + }; + + if (this.scheduler) { + 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, + }); + + return () => { + abortController.abort(); + }; + } + + const intervalKey = setInterval(runOnce, this.orphanCleanupIntervalMs); + return () => { + clearInterval(intervalKey); + }; } } @@ -290,9 +374,7 @@ function progressTracker() { const meter = metrics.getMeter('default'); const stitchedEntities = meter.createCounter( 'catalog.stitched.entities.count', - { - description: 'Amount of entities stitched', - }, + { description: 'Amount of entities stitched' }, ); const processedEntities = meter.createCounter( diff --git a/plugins/catalog-backend/src/service/CatalogBuilder.ts b/plugins/catalog-backend/src/service/CatalogBuilder.ts index 782ad1bf3c..a76913512f 100644 --- a/plugins/catalog-backend/src/service/CatalogBuilder.ts +++ b/plugins/catalog-backend/src/service/CatalogBuilder.ts @@ -15,6 +15,7 @@ */ import { PluginDatabaseManager, UrlReader } from '@backstage/backend-common'; +import { PluginTaskScheduler } from '@backstage/backend-tasks'; import { DefaultNamespaceEntityPolicy, Entity, @@ -115,6 +116,7 @@ export type CatalogEnvironment = { config: Config; reader: UrlReader; permissions: PermissionEvaluator | PermissionAuthorizer; + scheduler?: PluginTaskScheduler; }; /** @@ -424,7 +426,7 @@ export class CatalogBuilder { processingEngine: CatalogProcessingEngine; router: Router; }> { - const { config, database, logger, permissions } = this.env; + const { config, database, logger, permissions, scheduler } = this.env; const policy = this.buildEntityPolicy(); const processors = this.buildProcessors(); @@ -517,17 +519,19 @@ export class CatalogBuilder { provider => provider.getProviderName(), ); - const processingEngine = new DefaultCatalogProcessingEngine( + const processingEngine = new DefaultCatalogProcessingEngine({ + config, + scheduler, logger, processingDatabase, orchestrator, stitcher, - () => createHash('sha1'), - 1000, - event => { + createHash: () => createHash('sha1'), + pollingIntervalMs: 1000, + onProcessingError: event => { this.onProcessingError?.(event); }, - ); + }); const locationAnalyzer = this.locationAnalyzer ?? diff --git a/plugins/catalog-backend/src/service/CatalogPlugin.ts b/plugins/catalog-backend/src/service/CatalogPlugin.ts index a6b164001c..e0399c1683 100644 --- a/plugins/catalog-backend/src/service/CatalogPlugin.ts +++ b/plugins/catalog-backend/src/service/CatalogPlugin.ts @@ -76,6 +76,7 @@ export const catalogPlugin = createBackendPlugin({ database: coreServices.database, httpRouter: coreServices.httpRouter, lifecycle: coreServices.lifecycle, + scheduler: coreServices.scheduler, }, async init({ logger, @@ -85,6 +86,7 @@ export const catalogPlugin = createBackendPlugin({ permissions, httpRouter, lifecycle, + scheduler, }) { const winstonLogger = loggerToWinstonLogger(logger); const builder = await CatalogBuilder.create({ @@ -92,6 +94,7 @@ export const catalogPlugin = createBackendPlugin({ reader, permissions, database, + scheduler, logger: winstonLogger, }); builder.addProcessor(...processingExtensions.processors); diff --git a/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts b/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts index 85cec596ce..76284ebbfd 100644 --- a/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts +++ b/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts @@ -33,6 +33,7 @@ import { DefaultCatalogProcessingEngine } from '../processing/DefaultCatalogProc import { EntityProcessingRequest } from '../processing/types'; import { Stitcher } from '../stitching/Stitcher'; import { DefaultRefreshService } from './DefaultRefreshService'; +import { ConfigReader } from '@backstage/config'; jest.setTimeout(60_000); @@ -108,10 +109,11 @@ describe('DefaultRefreshService', () => { } } - const engine = new DefaultCatalogProcessingEngine( - defaultLogger, - db, - { + const engine = new DefaultCatalogProcessingEngine({ + config: new ConfigReader({}), + logger: defaultLogger, + processingDatabase: db, + orchestrator: { async process(request: EntityProcessingRequest) { const entityRef = stringifyEntityRef(request.entity); const entity = entityMap.get(entityRef); @@ -149,10 +151,10 @@ describe('DefaultRefreshService', () => { }; }, }, - new Stitcher(knex, defaultLogger), - () => createHash('sha1'), - 50, - ); + stitcher: new Stitcher(knex, defaultLogger), + createHash: () => createHash('sha1'), + pollingIntervalMs: 50, + }); return engine; }; diff --git a/plugins/catalog-backend/src/service/createRouter.ts b/plugins/catalog-backend/src/service/createRouter.ts index 830e909272..35a6426ae8 100644 --- a/plugins/catalog-backend/src/service/createRouter.ts +++ b/plugins/catalog-backend/src/service/createRouter.ts @@ -52,6 +52,7 @@ import { } from './util'; import type { ApiRouter } from '@backstage/backend-openapi-utils'; import spec from '../schema/openapi'; +import { PluginTaskScheduler } from '@backstage/backend-tasks'; /** * Options used by {@link createRouter}. @@ -64,6 +65,7 @@ export interface RouterOptions { locationService: LocationService; orchestrator?: CatalogProcessingOrchestrator; refreshService?: RefreshService; + scheduler?: PluginTaskScheduler; logger: Logger; config: Config; permissionIntegrationRouter?: express.Router; diff --git a/yarn.lock b/yarn.lock index e133bcf9a0..048565f9cb 100644 --- a/yarn.lock +++ b/yarn.lock @@ -5436,6 +5436,7 @@ __metadata: "@backstage/backend-common": "workspace:^" "@backstage/backend-openapi-utils": "workspace:^" "@backstage/backend-plugin-api": "workspace:^" + "@backstage/backend-tasks": "workspace:^" "@backstage/backend-test-utils": "workspace:^" "@backstage/catalog-client": "workspace:^" "@backstage/catalog-model": "workspace:^"