diff --git a/.changeset/poor-clouds-ring.md b/.changeset/poor-clouds-ring.md new file mode 100644 index 0000000000..62ba3928a7 --- /dev/null +++ b/.changeset/poor-clouds-ring.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend': patch +--- + +Fixed a bug where entities provided without a location key would always replace existing entities, rather than updating them. diff --git a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts index f6fc7fa5fa..3dc5325178 100644 --- a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts @@ -784,7 +784,10 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { if (!oldRef) { // Add any entity that does not exist in the database toAdd.push(upsertItem); - } else if (oldRef.locationKey !== item.deferred.locationKey) { + } else if ( + (oldRef?.locationKey ?? undefined) !== + (item.deferred.locationKey ?? undefined) + ) { // Remove and then re-add any entity that exists, but with a different location key toRemove.push(item.ref); toAdd.push(upsertItem); diff --git a/plugins/catalog-backend/src/integration.test.ts b/plugins/catalog-backend/src/integration.test.ts new file mode 100644 index 0000000000..67b63428db --- /dev/null +++ b/plugins/catalog-backend/src/integration.test.ts @@ -0,0 +1,543 @@ +/* + * Copyright 2022 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 { Logger } from 'winston'; +import { ConfigReader } from '@backstage/config'; +import { JsonObject } from '@backstage/types'; +import { CatalogProcessingEngine, EntityProvider } from './index'; +import { DatabaseManager, getVoidLogger } from '@backstage/backend-common'; +import { PermissionEvaluator } from '@backstage/plugin-permission-common'; +import { + Entity, + EntityPolicies, + stringifyEntityRef, +} from '@backstage/catalog-model'; +import { defaultEntityDataParser } from './modules/util/parse'; +import { DefaultCatalogProcessingOrchestrator } from './processing/DefaultCatalogProcessingOrchestrator'; +import { applyDatabaseMigrations } from './database/migrations'; +import { DefaultProcessingDatabase } from './database/DefaultProcessingDatabase'; +import { ScmIntegrations } from '@backstage/integration'; +import { DefaultCatalogRulesEnforcer } from './ingestion/CatalogRules'; +import { Stitcher } from './stitching/Stitcher'; +import { DefaultEntitiesCatalog } from './service/DefaultEntitiesCatalog'; +import { DefaultCatalogProcessingEngine } from './processing/DefaultCatalogProcessingEngine'; +import { createHash } from 'crypto'; +import { DefaultRefreshService } from './service/DefaultRefreshService'; +import { connectEntityProviders } from './processing/connectEntityProviders'; +import { EntitiesCatalog } from './catalog/types'; +import { RefreshOptions, RefreshService } from './service/types'; +import { + CatalogProcessorEmit, + EntityProviderConnection, + LocationSpec, + processingResult, +} from '@backstage/plugin-catalog-node'; +import { RefreshStateItem } from './database/types'; + +const voidLogger = getVoidLogger(); + +class TestProvider implements EntityProvider { + #connection?: EntityProviderConnection; + + getProviderName(): string { + return 'test'; + } + + async connect(connection: EntityProviderConnection): Promise { + this.#connection = connection; + } + + getConnection() { + if (!this.#connection) { + throw new Error('Provider is not connected yet'); + } + return this.#connection; + } +} + +type ProgressTracker = NonNullable< + ConstructorParameters[7] +>; + +class ProxyProgressTracker implements ProgressTracker { + #inner: ProgressTracker; + + constructor(inner: ProgressTracker) { + this.#inner = inner; + } + + processStart(item: RefreshStateItem) { + return this.#inner.processStart(item, voidLogger); + } + + setTracker(tracker: ProgressTracker) { + this.#inner = tracker; + } +} + +class NoopProgressTracker implements ProgressTracker { + static emptyTracking = { + markFailed() {}, + markProcessorsCompleted() {}, + markSuccessfulWithChanges() {}, + markSuccessfulWithErrors() {}, + markSuccessfulWithNoChanges() {}, + }; + + processStart() { + return NoopProgressTracker.emptyTracking; + } +} + +class WaitingProgressTracker implements ProgressTracker { + #resolve: (errors: Record) => void; + #promise: Promise>; + #counts = new Map(); + #errors = new Map(); + #inFlight = new Array>(); + + constructor(private readonly entityRefs?: Set) { + let resolve: (errors: Record) => void; + this.#promise = new Promise>(_resolve => { + resolve = _resolve; + }); + this.#resolve = resolve!; + } + + processStart(item: RefreshStateItem) { + if (this.entityRefs && !this.entityRefs.has(item.entityRef)) { + return NoopProgressTracker.emptyTracking; + } + + let resolve: () => void; + this.#inFlight.push( + new Promise(_resolve => { + resolve = _resolve; + }), + ); + + const currentCount = this.#counts.get(item.id) ?? 0; + this.#counts.set(item.id, currentCount); + + const onDone = () => { + this.#counts.set(item.id, currentCount + 1); + + if (Array.from(this.#counts.values()).every(c => c >= 2)) { + this.#resolve(Object.fromEntries(this.#errors)); + } + }; + return { + markFailed: (error: Error) => { + this.#errors.set(item.entityRef, error); + onDone(); + resolve(); + }, + markProcessorsCompleted() {}, + markSuccessfulWithChanges: () => { + this.#errors.delete(item.entityRef); + this.#counts.set(item.id, 0); + resolve(); + }, + markSuccessfulWithErrors: () => { + this.#errors.delete(item.entityRef); + onDone(); + resolve(); + }, + markSuccessfulWithNoChanges: () => { + onDone(); + resolve(); + }, + }; + } + + async wait(): Promise> { + return this.#promise; + } + + async waitForFinish(): Promise { + await Promise.all(this.#inFlight.slice()); + } +} + +class TestHarness { + readonly #catalog: EntitiesCatalog; + readonly #engine: CatalogProcessingEngine; + readonly #refresh: RefreshService; + readonly #provider: TestProvider; + readonly #proxyProgressTracker: ProxyProgressTracker; + + static async create(options?: { + config?: JsonObject; + logger?: Logger; + db?: Knex; + permissions?: PermissionEvaluator; + processEntity?( + entity: Entity, + location: LocationSpec, + emit: CatalogProcessorEmit, + ): Promise; + onProcessingError?(event: { + unprocessedEntity: Entity; + errors: Error[]; + }): void; + }) { + const config = new ConfigReader( + options?.config ?? { + backend: { + database: { + client: 'better-sqlite3', + connection: ':memory:', + }, + }, + }, + ); + const logger = options?.logger ?? getVoidLogger(); + const db = + options?.db ?? + (await DatabaseManager.fromConfig(config, { logger }) + .forPlugin('catalog') + .getClient()); + + await applyDatabaseMigrations(db); + + const processingDatabase = new DefaultProcessingDatabase({ + database: db, + logger, + refreshInterval: () => 0.05, + }); + + const integrations = ScmIntegrations.fromConfig(config); + const rulesEnforcer = DefaultCatalogRulesEnforcer.fromConfig(config); + const orchestrator = new DefaultCatalogProcessingOrchestrator({ + processors: [ + { + getProcessorName: () => 'test', + async validateEntityKind() { + return true; + }, + async preProcessEntity( + entity: Entity, + location: LocationSpec, + emit: CatalogProcessorEmit, + ) { + if (options?.processEntity) { + return options?.processEntity(entity, location, emit); + } + return entity; + }, + }, + ], + integrations, + rulesEnforcer, + logger, + parser: defaultEntityDataParser, + policy: EntityPolicies.allOf([]), + }); + const stitcher = new Stitcher(db, logger); + const catalog = new DefaultEntitiesCatalog(db, stitcher); + const proxyProgressTracker = new ProxyProgressTracker( + new NoopProgressTracker(), + ); + + const engine = new DefaultCatalogProcessingEngine( + logger, + processingDatabase, + orchestrator, + stitcher, + () => createHash('sha1'), + 50, + event => { + if (options?.onProcessingError) { + options.onProcessingError(event); + } else { + throw new Error( + `Catalog processing error, ${event.errors.join(', ')}`, + ); + } + }, + proxyProgressTracker, + ); + + const refresh = new DefaultRefreshService({ database: processingDatabase }); + + const provider = new TestProvider(); + + await connectEntityProviders(processingDatabase, [provider]); + + return new TestHarness( + catalog, + engine, + refresh, + provider, + proxyProgressTracker, + ); + } + + constructor( + catalog: EntitiesCatalog, + engine: CatalogProcessingEngine, + refresh: RefreshService, + provider: TestProvider, + proxyProgressTracker: ProxyProgressTracker, + ) { + this.#catalog = catalog; + this.#engine = engine; + this.#refresh = refresh; + this.#provider = provider; + this.#proxyProgressTracker = proxyProgressTracker; + } + + async process(entityRefs?: Set) { + const tracker = new WaitingProgressTracker(entityRefs); + this.#proxyProgressTracker.setTracker(tracker); + + this.#engine.start(); + + const errors = await tracker.wait(); + + this.#engine.stop(); + await tracker.waitForFinish(); + + this.#proxyProgressTracker.setTracker(new NoopProgressTracker()); + + return errors; + } + + async setInputEntities(entities: (Entity & { locationKey?: string })[]) { + return this.#provider.getConnection().applyMutation({ + type: 'full', + entities: entities.map(({ locationKey, ...entity }) => ({ + entity, + locationKey, + })), + }); + } + + async getOutputEntities(): Promise> { + const { entities } = await this.#catalog.entities(); + return Object.fromEntries(entities.map(e => [stringifyEntityRef(e), e])); + } + + async refresh(options: RefreshOptions) { + return this.#refresh.refresh(options); + } +} + +describe('Catalog Backend Integration', () => { + it('should add entities and update errors', async () => { + let triggerError = false; + + const harness = await TestHarness.create({ + async processEntity(entity: Entity) { + if (triggerError) { + throw new Error('NOPE'); + } + return entity; + }, + }); + + await harness.setInputEntities([ + { + apiVersion: 'backstage.io/v1alpha1', + kind: 'Component', + metadata: { + name: 'test', + annotations: { + 'backstage.io/managed-by-location': 'url:.', + 'backstage.io/managed-by-origin-location': 'url:.', + }, + }, + }, + ]); + + await expect(harness.getOutputEntities()).resolves.toEqual({}); + await expect(harness.process()).resolves.toEqual({}); + + await expect(harness.getOutputEntities()).resolves.toEqual({ + 'component:default/test': { + apiVersion: 'backstage.io/v1alpha1', + kind: 'Component', + metadata: expect.objectContaining({ name: 'test' }), + relations: [], + }, + }); + + triggerError = true; + + await expect(harness.process()).resolves.toEqual({}); + + await expect(harness.getOutputEntities()).resolves.toEqual({ + 'component:default/test': { + apiVersion: 'backstage.io/v1alpha1', + kind: 'Component', + metadata: expect.objectContaining({ name: 'test' }), + relations: [], + status: { + items: [ + { + level: 'error', + type: 'backstage.io/catalog-processing', + message: + 'InputError: Processor Object threw an error while preprocessing; caused by Error: NOPE', + error: { + name: 'InputError', + message: + 'Processor Object threw an error while preprocessing; caused by Error: NOPE', + cause: { + name: 'Error', + message: 'NOPE', + stack: expect.stringMatching(/^Error: NOPE/), + }, + }, + }, + ], + }, + }, + }); + }); + + it('should orphan entities', async () => { + const generatedApis = ['api-1', 'api-2']; + + const harness = await TestHarness.create({ + async processEntity( + entity: Entity, + location: LocationSpec, + emit: CatalogProcessorEmit, + ) { + if (entity.metadata.name === 'test') { + for (const api of generatedApis) { + emit( + processingResult.entity(location, { + apiVersion: 'backstage.io/v1alpha1', + kind: 'API', + metadata: { + name: api, + annotations: { + 'backstage.io/managed-by-location': 'url:.', + 'backstage.io/managed-by-origin-location': 'url:.', + }, + }, + }), + ); + } + } + return entity; + }, + }); + + await harness.setInputEntities([ + { + apiVersion: 'backstage.io/v1alpha1', + kind: 'Component', + metadata: { + name: 'test', + annotations: { + 'backstage.io/managed-by-location': 'url:.', + 'backstage.io/managed-by-origin-location': 'url:.', + }, + }, + }, + ]); + + await expect(harness.getOutputEntities()).resolves.toEqual({}); + await expect(harness.process()).resolves.toEqual({}); + + await expect(harness.getOutputEntities()).resolves.toEqual({ + 'component:default/test': { + apiVersion: 'backstage.io/v1alpha1', + kind: 'Component', + metadata: expect.objectContaining({ name: 'test' }), + relations: [], + }, + 'api:default/api-1': expect.objectContaining({ + metadata: expect.objectContaining({ name: 'api-1' }), + }), + 'api:default/api-2': expect.objectContaining({ + metadata: expect.objectContaining({ name: 'api-2' }), + }), + }); + + generatedApis.pop(); + + await expect(harness.process()).resolves.toEqual({}); + + await expect(harness.getOutputEntities()).resolves.toEqual({ + 'component:default/test': { + apiVersion: 'backstage.io/v1alpha1', + kind: 'Component', + metadata: expect.objectContaining({ name: 'test' }), + relations: [], + }, + 'api:default/api-1': expect.objectContaining({ + metadata: expect.objectContaining({ name: 'api-1' }), + }), + 'api:default/api-2': expect.objectContaining({ + metadata: expect.objectContaining({ + name: 'api-2', + annotations: expect.objectContaining({ + 'backstage.io/orphan': 'true', + }), + }), + }), + }); + }); + + it('should not replace matching provided entities', async () => { + const harness = await TestHarness.create(); + + const entityA = { + apiVersion: 'backstage.io/v1alpha1', + kind: 'Component', + metadata: { + name: 'a', + annotations: { + 'backstage.io/managed-by-location': 'url:.', + 'backstage.io/managed-by-origin-location': 'url:.', + }, + }, + }; + const entityB = { + apiVersion: 'backstage.io/v1alpha1', + kind: 'Component', + metadata: { + name: 'b', + annotations: { + 'backstage.io/managed-by-location': 'url:.', + 'backstage.io/managed-by-origin-location': 'url:.', + }, + }, + }; + + const entities = [entityA, { locationKey: 'loc', ...entityB }]; + + await harness.setInputEntities(entities); + await expect(harness.process()).resolves.toEqual({}); + + const outputEntities = await harness.getOutputEntities(); + + await expect(harness.getOutputEntities()).resolves.toEqual({ + 'component:default/a': expect.anything(), + 'component:default/b': expect.anything(), + }); + + await harness.setInputEntities(entities); + await expect(harness.process()).resolves.toEqual({}); + + await expect(harness.getOutputEntities()).resolves.toEqual(outputEntities); + }); +}); diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts index fb9518c783..4693ffc559 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts @@ -32,7 +32,6 @@ import { startTaskPipeline } from './TaskPipeline'; const CACHE_TTL = 5; export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { - private readonly tracker = progressTracker(); private stopFunc?: () => void; constructor( @@ -46,6 +45,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { unprocessedEntity: Entity; errors: Error[]; }) => Promise | void, + private readonly tracker = progressTracker(), ) {} async start() {