diff --git a/.changeset/metal-donuts-cross.md b/.changeset/metal-donuts-cross.md new file mode 100644 index 0000000000..6c2699ed61 --- /dev/null +++ b/.changeset/metal-donuts-cross.md @@ -0,0 +1,8 @@ +--- +'@backstage/plugin-catalog-backend': patch +--- + +This fixes a bug where locations couldn't be added unless the processing engine is started. +It's now possible to run the catalog backend without starting the processing engine and still allowing locations registrations. + +This is done by refactor the `EntityProvider.connect` to happen outside the engine. diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts index 1ce32a01e1..fb42af778b 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts @@ -14,11 +14,7 @@ * limitations under the License. */ -import { - Entity, - entityEnvelopeSchemaValidator, - stringifyEntityRef, -} from '@backstage/catalog-model'; +import { stringifyEntityRef } from '@backstage/catalog-model'; import { assertError, serializeError } from '@backstage/errors'; import { Hash } from 'crypto'; import stableStringify from 'fast-json-stable-stringify'; @@ -32,68 +28,15 @@ import { } from '../processing/types'; import { Stitcher } from '../stitching/Stitcher'; import { startTaskPipeline } from './TaskPipeline'; -import { - EntityProvider, - EntityProviderConnection, - EntityProviderMutation, -} from '../providers/types'; const CACHE_TTL = 5; -class Connection implements EntityProviderConnection { - readonly validateEntityEnvelope = entityEnvelopeSchemaValidator(); - - constructor( - private readonly config: { - id: string; - processingDatabase: ProcessingDatabase; - }, - ) {} - - async applyMutation(mutation: EntityProviderMutation): Promise { - const db = this.config.processingDatabase; - - if (mutation.type === 'full') { - this.check(mutation.entities.map(e => e.entity)); - await db.transaction(async tx => { - await db.replaceUnprocessedEntities(tx, { - sourceKey: this.config.id, - type: 'full', - items: mutation.entities, - }); - }); - } 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, - }); - }); - } - } - - private check(entities: Entity[]) { - for (const entity of entities) { - try { - this.validateEntityEnvelope(entity); - } catch (e) { - throw new TypeError(`Malformed entity envelope, ${e}`); - } - } - } -} - export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { private readonly tracker = progressTracker(); private stopFunc?: () => void; constructor( private readonly logger: Logger, - private readonly entityProviders: EntityProvider[], private readonly processingDatabase: ProcessingDatabase, private readonly orchestrator: CatalogProcessingOrchestrator, private readonly stitcher: Stitcher, @@ -106,15 +49,6 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { throw new Error('Processing engine is already started'); } - for (const provider of this.entityProviders) { - await provider.connect( - new Connection({ - id: provider.getProviderName(), - processingDatabase: this.processingDatabase, - }), - ); - } - this.stopFunc = startTaskPipeline({ lowWatermark: 5, highWatermark: 10, diff --git a/plugins/catalog-backend/src/processing/connectEntityProviders.ts b/plugins/catalog-backend/src/processing/connectEntityProviders.ts new file mode 100644 index 0000000000..02bdc2aee3 --- /dev/null +++ b/plugins/catalog-backend/src/processing/connectEntityProviders.ts @@ -0,0 +1,88 @@ +/* + * 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 { + Entity, + entityEnvelopeSchemaValidator, +} from '@backstage/catalog-model'; +import { ProcessingDatabase } from '../database/types'; +import { + EntityProvider, + EntityProviderConnection, + EntityProviderMutation, +} from '../providers/types'; + +class Connection implements EntityProviderConnection { + readonly validateEntityEnvelope = entityEnvelopeSchemaValidator(); + + constructor( + private readonly config: { + id: string; + processingDatabase: ProcessingDatabase; + }, + ) {} + + async applyMutation(mutation: EntityProviderMutation): Promise { + const db = this.config.processingDatabase; + + if (mutation.type === 'full') { + this.check(mutation.entities.map(e => e.entity)); + await db.transaction(async tx => { + await db.replaceUnprocessedEntities(tx, { + sourceKey: this.config.id, + type: 'full', + items: mutation.entities, + }); + }); + } 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, + }); + }); + } + } + + private check(entities: Entity[]) { + for (const entity of entities) { + try { + this.validateEntityEnvelope(entity); + } catch (e) { + throw new TypeError(`Malformed entity envelope, ${e}`); + } + } + } +} + +export async function connectEntityProviders( + db: ProcessingDatabase, + providers: EntityProvider[], +) { + await Promise.all( + providers.map(async provider => { + const connection = new Connection({ + id: provider.getProviderName(), + processingDatabase: db, + }); + return provider.connect(connection); + }), + ); +} diff --git a/plugins/catalog-backend/src/service/NextCatalogBuilder.ts b/plugins/catalog-backend/src/service/NextCatalogBuilder.ts index bbda6f070c..eed420cd53 100644 --- a/plugins/catalog-backend/src/service/NextCatalogBuilder.ts +++ b/plugins/catalog-backend/src/service/NextCatalogBuilder.ts @@ -80,6 +80,7 @@ import { DefaultCatalogRulesEnforcer } from '../ingestion/CatalogRules'; import { Config } from '@backstage/config'; import { Logger } from 'winston'; import { LocationService } from './types'; +import { connectEntityProviders } from '../processing/connectEntityProviders'; export type CatalogEnvironment = { logger: Logger; @@ -364,7 +365,6 @@ export class NextCatalogBuilder { const processingEngine = new DefaultCatalogProcessingEngine( logger, - entityProviders, processingDatabase, orchestrator, stitcher, @@ -390,6 +390,8 @@ export class NextCatalogBuilder { config, }); + await connectEntityProviders(processingDatabase, entityProviders); + return { entitiesCatalog, locationsCatalog,