diff --git a/.changeset/rude-windows-live.md b/.changeset/rude-windows-live.md new file mode 100644 index 0000000000..b8f0e0e158 --- /dev/null +++ b/.changeset/rude-windows-live.md @@ -0,0 +1,17 @@ +--- +'@backstage/plugin-catalog-backend': minor +--- + +Handle entity name conflicts in a deterministic way and avoid crashes due to naming conflicts at startup. + +This is a breaking change for the database and entity provider interfaces of the new catalog. The interfaces with breaking changes are `EntityProvider` and `ProcessingDatabase`, and while it's unlikely that these interfaces have much usage yet, a migration guide is provided below. + +The breaking change to the `EntityProvider` interface lies within the items passed in the `EntityProviderMutation` type. Rather than passing along entities directly, they are now wrapped up in a `DeferredEntity` type, which is a tuple of an `entity` and a `locationKey`. The `entity` houses the entity as it was passed on before, while the `locationKey` is a new concept that is used for conflict resolution within the catalog. + +The `locationKey` is an opaque string that should be unique for each location that an entity could be located at, and undefined if the entity does not have a fixed location. In practice it should be set to the serialized location reference if the entity is stored in Git, for example `https://github.com/backstage/backstage/blob/master/catalog-info.yaml`. A conflict between two entity definitions happen when they have the same entity reference, i.e. kind, namespace, and name. In the event of a conflict the location key will be used according to the following rules to resolve the conflict: + +- If the entity is already present in the database but does not have a location key set, the new entity wins and will override the existing one. +- If the entity is already present in the database the new entity will only win if the location keys of the existing and new entity are the same. +- If the entity is not already present, insert the entity into the database along with the provided location key. + +The breaking change to the `ProcessingDatabase` is similar to the one for the entity provider, as it reflects the switch from `Entity` to `DeferredEntity` in the `ReplaceUnprocessedEntitiesOptions`. In addition, the `addUnprocessedEntities` method has been removed from the `ProcessingDatabase` interface, and the `RefreshStateItem` and `UpdateProcessedEntityOptions` types have received a new optional `locationKey` property. diff --git a/plugins/catalog-backend/migrations/20210622104022_refresh_state_location_key.js b/plugins/catalog-backend/migrations/20210622104022_refresh_state_location_key.js new file mode 100644 index 0000000000..37ab5e12cb --- /dev/null +++ b/plugins/catalog-backend/migrations/20210622104022_refresh_state_location_key.js @@ -0,0 +1,40 @@ +/* + * Copyright 2020 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. + */ + +// @ts-check + +/** + * @param {import('knex').Knex} knex + */ +exports.up = async function up(knex) { + await knex.schema.alterTable('refresh_state', table => { + table + .text('location_key') + .nullable() + .comment( + 'An opaque key that uniquely identifies the location of an entity in order to support conflict resolution', + ); + }); +}; + +/** + * @param {import('knex').Knex} knex + */ +exports.down = async function down(knex) { + await knex.schema.alterTable('refresh_state', table => { + table.dropColumn('location_key'); + }); +}; diff --git a/plugins/catalog-backend/src/next/ConfigLocationEntityProvider.test.ts b/plugins/catalog-backend/src/next/ConfigLocationEntityProvider.test.ts index 37d549c7f2..7592182bad 100644 --- a/plugins/catalog-backend/src/next/ConfigLocationEntityProvider.test.ts +++ b/plugins/catalog-backend/src/next/ConfigLocationEntityProvider.test.ts @@ -41,26 +41,34 @@ describe('ConfigLocationEntityProvider', () => { expect(mockConnection.applyMutation).toHaveBeenCalledWith({ type: 'full', entities: expect.arrayContaining([ - expect.objectContaining({ - spec: { - target: path.join( - resolvePackagePath('@backstage/plugin-catalog-backend'), - './lols.yaml', - ), - type: 'file', - }, - }), + { + entity: expect.objectContaining({ + spec: { + target: path.join( + resolvePackagePath('@backstage/plugin-catalog-backend'), + './lols.yaml', + ), + type: 'file', + }, + }), + locationKey: expect.stringMatching( + /plugins\/catalog-backend\/lols\.yaml$/, + ), + }, ]), }); expect(mockConnection.applyMutation).toHaveBeenCalledWith({ type: 'full', entities: expect.arrayContaining([ - expect.objectContaining({ - spec: { - target: 'https://github.com/backstage/backstage', - type: 'url', - }, - }), + { + entity: expect.objectContaining({ + spec: { + target: 'https://github.com/backstage/backstage', + type: 'url', + }, + }), + locationKey: 'url:https://github.com/backstage/backstage', + }, ]), }); }); diff --git a/plugins/catalog-backend/src/next/ConfigLocationEntityProvider.ts b/plugins/catalog-backend/src/next/ConfigLocationEntityProvider.ts index 9218a9c0e1..71ec84dae7 100644 --- a/plugins/catalog-backend/src/next/ConfigLocationEntityProvider.ts +++ b/plugins/catalog-backend/src/next/ConfigLocationEntityProvider.ts @@ -16,6 +16,7 @@ import { Config } from '@backstage/config'; import path from 'path'; +import { getEntityLocationRef } from './processing/util'; import { EntityProvider, EntityProviderConnection } from './types'; import { locationSpecToLocationEntity } from './util'; @@ -37,10 +38,12 @@ export class ConfigLocationEntityProvider implements EntityProvider { const entities = locationConfigs.map(location => { const type = location.getString('type'); const target = location.getString('target'); - return locationSpecToLocationEntity({ + const entity = locationSpecToLocationEntity({ type, target: type === 'file' ? path.resolve(target) : target, }); + const locationKey = getEntityLocationRef(entity); + return { entity, locationKey }; }); await this.connection.applyMutation({ diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index a504de0bf3..0ae20c94b1 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -46,7 +46,7 @@ class Connection implements EntityProviderConnection { const db = this.config.processingDatabase; if (mutation.type === 'full') { - this.check(mutation.entities); + this.check(mutation.entities.map(e => e.entity)); await db.transaction(async tx => { await db.replaceUnprocessedEntities(tx, { sourceKey: this.config.id, @@ -57,8 +57,8 @@ class Connection implements EntityProviderConnection { return; } - this.check(mutation.added); - this.check(mutation.removed); + 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, @@ -125,7 +125,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { }, processTask: async item => { try { - const { id, state, unprocessedEntity, entityRef } = item; + const { id, state, unprocessedEntity, entityRef, locationKey } = item; const result = await this.orchestrator.process({ entity: unprocessedEntity, state, @@ -171,6 +171,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { errors: errorsString, relations: result.relations, deferredEntities: result.deferredEntities, + locationKey, }); }); diff --git a/plugins/catalog-backend/src/next/DefaultLocationService.test.ts b/plugins/catalog-backend/src/next/DefaultLocationService.test.ts index ae2c4e9a20..f965ec3077 100644 --- a/plugins/catalog-backend/src/next/DefaultLocationService.test.ts +++ b/plugins/catalog-backend/src/next/DefaultLocationService.test.ts @@ -48,11 +48,14 @@ describe('DefaultLocationServiceTest', () => { }, deferredEntities: [ { - apiVersion: 'backstage.io/v1alpha1', - kind: 'Component', - metadata: { - name: 'bar', + entity: { + apiVersion: 'backstage.io/v1alpha1', + kind: 'Component', + metadata: { + name: 'bar', + }, }, + locationKey: 'file:///tmp/mock.yaml', }, ], relations: [], diff --git a/plugins/catalog-backend/src/next/DefaultLocationService.ts b/plugins/catalog-backend/src/next/DefaultLocationService.ts index ceb51a952e..2820ba14b1 100644 --- a/plugins/catalog-backend/src/next/DefaultLocationService.ts +++ b/plugins/catalog-backend/src/next/DefaultLocationService.ts @@ -20,7 +20,10 @@ import { LOCATION_ANNOTATION, ORIGIN_LOCATION_ANNOTATION, } from '@backstage/catalog-model'; -import { CatalogProcessingOrchestrator } from './processing/types'; +import { + CatalogProcessingOrchestrator, + DeferredEntity, +} from './processing/types'; import { LocationService, LocationStore } from './types'; import { locationSpecToMetadataName } from './util'; @@ -73,7 +76,9 @@ export class DefaultLocationService implements LocationService { target: spec.target, }, }; - const unprocessedEntities: Entity[] = [entity]; + const unprocessedEntities: DeferredEntity[] = [ + { entity, locationKey: `${spec.type}:${spec.target}` }, + ]; const entities: Entity[] = []; const state = new Map(); // ignored while (unprocessedEntities.length) { @@ -82,7 +87,7 @@ export class DefaultLocationService implements LocationService { continue; } const processed = await this.orchestrator.process({ - entity: currentEntity, + entity: currentEntity.entity, state, }); diff --git a/plugins/catalog-backend/src/next/DefaultLocationStore.test.ts b/plugins/catalog-backend/src/next/DefaultLocationStore.test.ts index 762a585754..82b08f2e70 100644 --- a/plugins/catalog-backend/src/next/DefaultLocationStore.test.ts +++ b/plugins/catalog-backend/src/next/DefaultLocationStore.test.ts @@ -113,13 +113,17 @@ describe('DefaultLocationStore', () => { type: 'delta', removed: [], added: expect.arrayContaining([ - expect.objectContaining({ - spec: { - target: - 'https://github.com/backstage/demo/blob/master/catalog-info.yml', - type: 'url', - }, - }), + { + entity: expect.objectContaining({ + spec: { + target: + 'https://github.com/backstage/demo/blob/master/catalog-info.yml', + type: 'url', + }, + }), + locationKey: + 'url:https://github.com/backstage/demo/blob/master/catalog-info.yml', + }, ]), }); }, @@ -156,15 +160,19 @@ describe('DefaultLocationStore', () => { expect(connection.applyMutation).toHaveBeenCalledWith({ type: 'delta', added: [], - removed: expect.arrayContaining([ - expect.objectContaining({ - spec: { - target: - 'https://github.com/backstage/demo/blob/master/catalog-info.yml', - type: 'url', - }, - }), - ]), + removed: [ + { + entity: expect.objectContaining({ + spec: { + target: + 'https://github.com/backstage/demo/blob/master/catalog-info.yml', + type: 'url', + }, + }), + locationKey: + 'url:https://github.com/backstage/demo/blob/master/catalog-info.yml', + }, + ], }); }, 60_000, diff --git a/plugins/catalog-backend/src/next/DefaultLocationStore.ts b/plugins/catalog-backend/src/next/DefaultLocationStore.ts index 09de77e49f..c9ffe9a682 100644 --- a/plugins/catalog-backend/src/next/DefaultLocationStore.ts +++ b/plugins/catalog-backend/src/next/DefaultLocationStore.ts @@ -19,6 +19,7 @@ import { ConflictError, NotFoundError } from '@backstage/errors'; import { Knex } from 'knex'; import { v4 as uuid } from 'uuid'; import { DbLocationsRow } from './database/tables'; +import { getEntityLocationRef } from './processing/util'; import { EntityProvider, EntityProviderConnection, @@ -60,10 +61,10 @@ export class DefaultLocationStore implements LocationStore, EntityProvider { return inner; }); - + const entity = locationSpecToLocationEntity(location); await this.connection.applyMutation({ type: 'delta', - added: [locationSpecToLocationEntity(location)], + added: [{ entity, locationKey: getEntityLocationRef(entity) }], removed: [], }); @@ -102,11 +103,11 @@ export class DefaultLocationStore implements LocationStore, EntityProvider { await tx('locations').where({ id }).del(); return location; }); - + const entity = locationSpecToLocationEntity(deleted); await this.connection.applyMutation({ type: 'delta', added: [], - removed: [locationSpecToLocationEntity(deleted)], + removed: [{ entity, locationKey: getEntityLocationRef(entity) }], }); } @@ -124,7 +125,8 @@ export class DefaultLocationStore implements LocationStore, EntityProvider { const locations = await this.locations(); const entities = locations.map(location => { - return locationSpecToLocationEntity(location); + const entity = locationSpecToLocationEntity(location); + return { entity, locationKey: getEntityLocationRef(entity) }; }); await this.connection.applyMutation({ diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts index a0db3b293f..e449e5580c 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts @@ -19,6 +19,7 @@ import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils'; import { Entity, stringifyEntityRef } from '@backstage/catalog-model'; import { JsonObject } from '@backstage/config'; import { Knex } from 'knex'; +import { Logger } from 'winston'; import * as uuid from 'uuid'; import { DatabaseManager } from './DatabaseManager'; import { DefaultProcessingDatabase } from './DefaultProcessingDatabase'; @@ -29,12 +30,15 @@ import { } from './tables'; describe('Default Processing Database', () => { - const logger = getVoidLogger(); + const defaultLogger = getVoidLogger(); const databases = TestDatabases.create({ ids: ['POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'], }); - async function createDatabase(databaseId: TestDatabaseId) { + async function createDatabase( + databaseId: TestDatabaseId, + logger: Logger = defaultLogger, + ) { const knex = await databases.init(databaseId); await DatabaseManager.createDatabase(knex); return { @@ -57,6 +61,123 @@ describe('Default Processing Database', () => { await db('refresh_state').insert(ref); }; + describe('addUprocessedEntities', () => { + function mockEntity(name: string, type: string): Entity { + return { + apiVersion: '1', + kind: 'Component', + metadata: { + name, + }, + spec: { + type, + }, + }; + } + + it.each(databases.eachSupportedId())( + 'updates refresh state with varying location keys, %p', + async databaseId => { + const mockWarn = jest.fn(); + const { db } = await createDatabase(databaseId, ({ + debug: jest.fn(), + warn: mockWarn, + } as unknown) as Logger); + await db.transaction(async tx => { + const knexTx = tx as Knex.Transaction; + + const steps = [ + { + locationKey: undefined, + expectedLocationKey: null, + type: 'a', + expectedType: 'a', + }, + { + locationKey: undefined, + expectedLocationKey: null, + type: 'b', + expectedType: 'b', + }, + { + locationKey: 'x', + expectedLocationKey: 'x', + type: 'c', + expectedType: 'c', + }, + { + locationKey: 'y', + expectedLocationKey: 'x', + type: 'd', + expectedType: 'c', + expectConflict: true, + }, + { + locationKey: undefined, + expectedLocationKey: 'x', + type: 'e', + expectedType: 'c', + expectConflict: true, + }, + { + locationKey: 'x', + expectedLocationKey: 'x', + type: 'f', + expectedType: 'f', + }, + ]; + for (const step of steps) { + mockWarn.mockClear(); + + await db.addUnprocessedEntities(tx, { + sourceKey: 'testing', + entities: [ + { + entity: mockEntity('1', step.type), + locationKey: step.locationKey, + }, + ], + }); + + if (step.expectConflict) { + // eslint-disable-next-line jest/no-conditional-expect + expect(mockWarn).toHaveBeenCalledWith( + expect.stringMatching(/^Detected conflicting entityRef/), + ); + } else { + // eslint-disable-next-line jest/no-conditional-expect + expect(mockWarn).not.toHaveBeenCalled(); + } + + const entities = await knexTx( + 'refresh_state', + ).select(); + expect(entities).toEqual([ + expect.objectContaining({ + entity_ref: 'component:default/1', + location_key: step.expectedLocationKey, + }), + ]); + const entity = JSON.parse(entities[0].unprocessed_entity) as Entity; + expect(entity.spec?.type).toEqual(step.expectedType); + + await expect( + knexTx( + 'refresh_state_references', + ).select(), + ).resolves.toEqual([ + expect.objectContaining({ + source_key: 'testing', + target_entity_ref: 'component:default/1', + }), + ]); + } + }); + }, + 60_000, + ); + }); + describe('updateProcessedEntity', () => { let id: string; let processedEntity: Entity; @@ -77,7 +198,7 @@ describe('Default Processing Database', () => { }); it.each(databases.eachSupportedId())( - 'fails when there is no processing state for the entity, %p', + 'fails when an entity is processed with a different locationKey, %p', async databaseId => { const { db } = await createDatabase(databaseId); await db.transaction(async tx => { @@ -89,12 +210,55 @@ describe('Default Processing Database', () => { relations: [], deferredEntities: [], }), - ).rejects.toThrow(`Processing state not found for ${id}`); + ).rejects.toThrow( + `Conflicting write of processing result for ${id} with location key 'undefined'`, + ); }); }, 60_000, ); + it.each(databases.eachSupportedId())( + 'fails when the locationKey is different, %p', + async databaseId => { + const options = { + id, + processedEntity, + state: new Map(), + relations: [], + deferredEntities: [], + locationKey: 'key', + errors: "['something broke']", + }; + const { knex, db } = await createDatabase(databaseId); + await insertRefreshStateRow(knex, { + entity_id: id, + entity_ref: 'location:default/fakelocation', + unprocessed_entity: '{}', + processed_entity: '{}', + location_key: 'key', + errors: '[]', + next_update_at: '2021-04-01 13:37:00', + last_discovery_at: '2021-04-01 13:37:00', + }); + await db.transaction(tx => db.updateProcessedEntity(tx, options)); + + const entities = await knex( + 'refresh_state', + ).select(); + expect(entities.length).toBe(1); + + await db.transaction(tx => + expect( + db.updateProcessedEntity(tx, { ...options, locationKey: 'fail' }), + ).rejects.toThrow( + `Conflicting write of processing result for ${id} with location key 'fail'`, + ), + ); + }, + 60_000, + ); + it.each(databases.eachSupportedId())( 'updates the refresh state entry with the cache, processed entity and errors, %p', async databaseId => { @@ -119,6 +283,7 @@ describe('Default Processing Database', () => { state, relations: [], deferredEntities: [], + locationKey: 'key', errors: "['something broke']", }), ); @@ -132,6 +297,7 @@ describe('Default Processing Database', () => { ); expect(entities[0].cache).toEqual(JSON.stringify(state)); expect(entities[0].errors).toEqual("['something broke']"); + expect(entities[0].location_key).toEqual('key'); }, 60_000, ); @@ -206,11 +372,14 @@ describe('Default Processing Database', () => { const deferredEntities = [ { - apiVersion: '1', - kind: 'Location', - metadata: { - name: 'next', + entity: { + apiVersion: '1', + kind: 'Location', + metadata: { + name: 'next', + }, }, + locationKey: 'mock', }, ]; @@ -227,7 +396,7 @@ describe('Default Processing Database', () => { const refreshStateEntries = await knex( 'refresh_state', ) - .where({ entity_ref: stringifyEntityRef(deferredEntities[0]) }) + .where({ entity_ref: stringifyEntityRef(deferredEntities[0].entity) }) .select(); expect(refreshStateEntries).toHaveLength(1); @@ -297,12 +466,15 @@ describe('Default Processing Database', () => { sourceKey: 'config', items: [ { - apiVersion: '1.0.0', - metadata: { - name: 'new-root', - }, - kind: 'Location', - } as Entity, + entity: { + apiVersion: '1.0.0', + metadata: { + name: 'new-root', + }, + kind: 'Location', + } as Entity, + locationKey: 'file:///tmp/foobar', + }, ], }), ); @@ -411,12 +583,15 @@ describe('Default Processing Database', () => { sourceKey: 'config', items: [ { - apiVersion: '1.0.0', - metadata: { - name: 'new-root', - }, - kind: 'Location', - } as Entity, + entity: { + apiVersion: '1.0.0', + metadata: { + name: 'new-root', + }, + kind: 'Location', + } as Entity, + locationKey: 'file:/tmp/foobar', + }, ], }); }); @@ -510,12 +685,15 @@ describe('Default Processing Database', () => { removed: [], added: [ { - apiVersion: '1.0.0', - metadata: { - name: 'new-root', - }, - kind: 'Location', - } as Entity, + entity: { + apiVersion: '1.0.0', + metadata: { + name: 'new-root', + }, + kind: 'Location', + } as Entity, + locationKey: 'file:///tmp/foobar', + }, ], }); }); @@ -614,12 +792,15 @@ describe('Default Processing Database', () => { added: [], removed: [ { - apiVersion: '1.0.0', - metadata: { - name: 'new-root', - }, - kind: 'Location', - } as Entity, + entity: { + apiVersion: '1.0.0', + metadata: { + name: 'new-root', + }, + kind: 'Location', + } as Entity, + locationKey: 'file:/tmp/foobar', + }, ], }); }); @@ -648,6 +829,73 @@ describe('Default Processing Database', () => { }, 60_000, ); + + it.each(databases.eachSupportedId())( + 'should update the location key during full replace, %p', + async databaseId => { + const { knex, db } = await createDatabase(databaseId); + await createLocations(knex, ['location:default/removed']); + await insertRefreshStateRow(knex, { + entity_id: uuid.v4(), + entity_ref: 'location:default/replaced', + unprocessed_entity: '{}', + processed_entity: '{}', + errors: '[]', + next_update_at: '2021-04-01 13:37:00', + last_discovery_at: '2021-04-01 13:37:00', + location_key: 'file:///tmp/old', + }); + + await insertRefRow(knex, { + source_key: 'lols', + target_entity_ref: 'location:default/removed', + }); + await insertRefRow(knex, { + source_key: 'lols', + target_entity_ref: 'location:default/replaced', + }); + + await db.transaction(async tx => { + await db.replaceUnprocessedEntities(tx, { + type: 'full', + sourceKey: 'lols', + items: [ + { + entity: { + apiVersion: '1.0.0', + metadata: { + name: 'replaced', + }, + kind: 'Location', + } as Entity, + locationKey: 'file:///tmp/foobar', + }, + ], + }); + }); + + const currentRefreshState = await knex( + 'refresh_state', + ).select(); + expect(currentRefreshState).toEqual([ + expect.objectContaining({ + entity_ref: 'location:default/replaced', + location_key: 'file:///tmp/foobar', + }), + ]); + + const currentRefRowState = await knex( + 'refresh_state_references', + ).select(); + expect(currentRefRowState).toEqual([ + expect.objectContaining({ + source_key: 'lols', + target_entity_ref: 'location:default/replaced', + }), + ]); + }, + 60_000, + ); }); describe('getProcessableEntities', () => { diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index 79daee13e6..1f1ea4bf84 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -16,12 +16,13 @@ import { Entity, stringifyEntityRef } from '@backstage/catalog-model'; import { JsonObject } from '@backstage/config'; -import { ConflictError, NotFoundError } from '@backstage/errors'; +import { ConflictError } from '@backstage/errors'; import { Knex } from 'knex'; import lodash from 'lodash'; import { v4 as uuid } from 'uuid'; import type { Logger } from 'winston'; import { Transaction } from '../../database'; +import { DeferredEntity } from '../processing/types'; import { DbRefreshStateReferencesRow, DbRefreshStateRow, @@ -63,23 +64,34 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { errors, relations, deferredEntities, + locationKey, } = options; - const refreshResult = await tx('refresh_state') .update({ processed_entity: JSON.stringify(processedEntity), cache: JSON.stringify(state), errors, + location_key: locationKey, }) - .where('entity_id', id); + .where('entity_id', id) + .andWhere(inner => { + if (!locationKey) { + return inner.whereNull('location_key'); + } + return inner + .where('location_key', locationKey) + .orWhereNull('location_key'); + }); if (refreshResult === 0) { - throw new NotFoundError(`Processing state not found for ${id}`); + throw new ConflictError( + `Conflicting write of processing result for ${id} with location key '${locationKey}'`, + ); } // Schedule all deferred entities for future processing. await this.addUnprocessedEntities(tx, { entities: deferredEntities, - entityRef: stringifyEntityRef(processedEntity), + sourceEntityRef: stringifyEntityRef(processedEntity), }); // Update fragments @@ -129,32 +141,51 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { private async createDelta( tx: Knex.Transaction, options: ReplaceUnprocessedEntitiesOptions, - ): Promise<{ toAdd: Entity[]; toRemove: string[] }> { + ): Promise<{ toAdd: DeferredEntity[]; toRemove: string[] }> { if (options.type === 'delta') { return { toAdd: options.added, - toRemove: options.removed.map(e => stringifyEntityRef(e)), + toRemove: options.removed.map(e => stringifyEntityRef(e.entity)), }; } + // Grab all of the existing references from the same source, and their locationKeys as well const oldRefs = await tx( 'refresh_state_references', ) .where({ source_key: options.sourceKey }) - .select('target_entity_ref') - .then(rows => rows.map(r => r.target_entity_ref)); + .leftJoin('refresh_state', { + target_entity_ref: 'entity_ref', + }) + .select(['target_entity_ref', 'location_key']); - const items = options.items.map(entity => ({ - entity, - ref: stringifyEntityRef(entity), + const items = options.items.map(deferred => ({ + deferred, + ref: stringifyEntityRef(deferred.entity), })); - const oldRefsSet = new Set(oldRefs); + const oldRefsSet = new Map( + oldRefs.map(r => [r.target_entity_ref, r.location_key]), + ); const newRefsSet = new Set(items.map(item => item.ref)); - const toAdd = items.filter(item => !oldRefsSet.has(item.ref)); - const toRemove = oldRefs.filter(ref => !newRefsSet.has(ref)); - return { toAdd: toAdd.map(({ entity }) => entity), toRemove }; + const toAdd = new Array(); + const toRemove = oldRefs + .map(row => row.target_entity_ref) + .filter(ref => !newRefsSet.has(ref)); + + for (const item of items) { + if (!oldRefsSet.has(item.ref)) { + // Add any entity that does not exist in the database + toAdd.push(item.deferred); + } else if (oldRefsSet.get(item.ref) !== item.deferred.locationKey) { + // Remove and then re-add any entity that exists, but with a different location key + toRemove.push(item.ref); + toAdd.push(item.deferred); + } + } + + return { toAdd, toRemove }; } async replaceUnprocessedEntities( @@ -281,28 +312,10 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { } if (toAdd.length) { - const state: Knex.DbRecord[] = toAdd.map(entity => ({ - entity_id: uuid(), - entity_ref: stringifyEntityRef(entity), - unprocessed_entity: JSON.stringify(entity), - errors: '', - next_update_at: tx.fn.now(), - last_discovery_at: tx.fn.now(), - })); - - const stateReferences: DbRefreshStateReferencesRow[] = toAdd.map( - entity => ({ - source_key: options.sourceKey, - target_entity_ref: stringifyEntityRef(entity), - }), - ); - // TODO(freben): Concurrency? If we did these one by one, a .onConflict().merge would have made sense - await tx.batchInsert('refresh_state', state, BATCH_SIZE); - await tx.batchInsert( - 'refresh_state_references', - stateReferences, - BATCH_SIZE, - ); + await this.addUnprocessedEntities(tx, { + sourceKey: options.sourceKey, + entities: toAdd, + }); } } @@ -312,44 +325,123 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { ): Promise { const tx = txOpaque as Knex.Transaction; - const stateRows = options.entities.map( - entity => - ({ - entity_id: uuid(), - entity_ref: stringifyEntityRef(entity), - unprocessed_entity: JSON.stringify(entity), - errors: '', - next_update_at: tx.fn.now(), - last_discovery_at: tx.fn.now(), - } as Knex.DbRecord), - ); - const stateReferenceRows = stateRows.map( - stateRow => - ({ - source_entity_ref: options.entityRef, - target_entity_ref: stateRow.entity_ref, - } as Knex.DbRecord), - ); + // Keeps track of the entities that we end up inserting to update refresh_state_references afterwards + const stateReferences = new Array(); + const conflictingStateReferences = new Array(); // Upsert all of the unprocessed entities into the refresh_state table, by // their entity ref. - // TODO(freben): Can this be batched somehow? - for (const row of stateRows) { - await tx('refresh_state') - .insert(row) - .onConflict('entity_ref') - .merge(['unprocessed_entity', 'last_discovery_at']); + for (const { entity, locationKey } of options.entities) { + const entityRef = stringifyEntityRef(entity); + const serializedEntity = JSON.stringify(entity); + + // We optimistically try to update any existing refresh state first, as this is by far + // the most common case. + const refreshResult = await tx('refresh_state') + .update({ + unprocessed_entity: serializedEntity, + location_key: locationKey, + last_discovery_at: tx.fn.now(), + }) + .where('entity_ref', entityRef) + .andWhere(inner => { + if (!locationKey) { + return inner.whereNull('location_key'); + } + return inner + .where('location_key', locationKey) + .orWhereNull('location_key'); + }); + + if (refreshResult === 0) { + // In the event that we can't update an existing refresh state, we first try to insert a new row + try { + let query = tx('refresh_state').insert({ + entity_id: uuid(), + entity_ref: entityRef, + unprocessed_entity: serializedEntity, + errors: '', + location_key: locationKey, + next_update_at: tx.fn.now(), + last_discovery_at: tx.fn.now(), + }); + + // TODO(Rugvip): only tested towards Postgres and SQLite + // We have to do this because the only way to detect if there was a conflict with + // SQLite is to catch the error, while Postgres needs to ignore the conflict to not + // break the ongoing transaction. + if (tx.client.config.client !== 'sqlite3') { + query = query.onConflict('entity_ref').ignore(); + } + + const result: { /* postgres */ rowCount?: number } = await query; + if (result.rowCount === 0) { + throw new ConflictError( + 'Insert failed due to conflicting entity_ref', + ); + } + } catch (error) { + if ( + !error.message.includes('UNIQUE constraint failed') && + error.name !== 'ConflictError' + ) { + throw error; + } + // If the row can't be inserted, we have a conflict, but it could be either + // because of a conflicting locationKey or a race with another instance, so check + // whether the conflicting entity has the same entityRef but a different locationKey + const [conflictingEntity] = await tx( + 'refresh_state', + ) + .where({ entity_ref: entityRef }) + .select(); + + // If the location key matches it means we just had a race trigger, which we can safely ignore + if ( + !conflictingEntity || + conflictingEntity.location_key !== locationKey + ) { + this.options.logger.warn( + `Detected conflicting entityRef ${entityRef} already referenced by ${conflictingEntity.location_key} and now also ${locationKey}`, + ); + conflictingStateReferences.push(entityRef); + continue; + } + } + } + + // Skipped on locationKey conflict + stateReferences.push(entityRef); } - // Replace all references for the originating entity before creating new ones - await tx('refresh_state_references') - .where({ source_entity_ref: options.entityRef }) - .delete(); - await tx.batchInsert( - 'refresh_state_references', - stateReferenceRows, - BATCH_SIZE, - ); + // Replace all references for the originating entity or source and then create new ones + if ('sourceKey' in options) { + await tx('refresh_state_references') + .whereNotIn('target_entity_ref', conflictingStateReferences) + .andWhere({ source_key: options.sourceKey }) + .delete(); + await tx.batchInsert( + 'refresh_state_references', + stateReferences.map(entityRef => ({ + source_key: options.sourceKey, + target_entity_ref: entityRef, + })), + BATCH_SIZE, + ); + } else { + await tx('refresh_state_references') + .whereNotIn('target_entity_ref', conflictingStateReferences) + .andWhere({ source_entity_ref: options.sourceEntityRef }) + .delete(); + await tx.batchInsert( + 'refresh_state_references', + stateReferences.map(entityRef => ({ + source_entity_ref: options.sourceEntityRef, + target_entity_ref: entityRef, + })), + BATCH_SIZE, + ); + } } async getProcessableEntities( @@ -406,6 +498,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { ? JSON.parse(i.cache) : new Map(), errors: i.errors, + locationKey: i.location_key, } as RefreshStateItem), ), }; diff --git a/plugins/catalog-backend/src/next/database/tables.ts b/plugins/catalog-backend/src/next/database/tables.ts index 93453ba914..395694131d 100644 --- a/plugins/catalog-backend/src/next/database/tables.ts +++ b/plugins/catalog-backend/src/next/database/tables.ts @@ -29,6 +29,7 @@ export type DbRefreshStateRow = { next_update_at: string; last_discovery_at: string; // remove? errors?: string; + location_key?: string; }; export type DbRefreshStateReferencesRow = { diff --git a/plugins/catalog-backend/src/next/database/types.ts b/plugins/catalog-backend/src/next/database/types.ts index dd342126c4..78199900e4 100644 --- a/plugins/catalog-backend/src/next/database/types.ts +++ b/plugins/catalog-backend/src/next/database/types.ts @@ -17,11 +17,17 @@ import { Entity, EntityRelationSpec } from '@backstage/catalog-model'; import { JsonObject } from '@backstage/config'; import { Transaction } from '../../database/types'; +import { DeferredEntity } from '../processing/types'; -export type AddUnprocessedEntitiesOptions = { - entityRef: string; - entities: Entity[]; -}; +export type AddUnprocessedEntitiesOptions = + | { + sourceEntityRef: string; + entities: DeferredEntity[]; + } + | { + sourceKey: string; + entities: DeferredEntity[]; + }; export type AddUnprocessedEntitiesResult = {}; @@ -31,7 +37,8 @@ export type UpdateProcessedEntityOptions = { state?: Map; errors?: string; relations: EntityRelationSpec[]; - deferredEntities: Entity[]; + deferredEntities: DeferredEntity[]; + locationKey?: string; }; export type UpdateProcessedEntityErrorsOptions = { @@ -48,6 +55,7 @@ export type RefreshStateItem = { lastDiscoveryAt: string; // remove? state: Map; errors?: string; + locationKey?: string; }; export type GetProcessableEntitiesResult = { @@ -57,24 +65,19 @@ export type GetProcessableEntitiesResult = { export type ReplaceUnprocessedEntitiesOptions = | { sourceKey: string; - items: Entity[]; + items: DeferredEntity[]; type: 'full'; } | { sourceKey: string; - added: Entity[]; - removed: Entity[]; + added: DeferredEntity[]; + removed: DeferredEntity[]; type: 'delta'; }; export interface ProcessingDatabase { transaction(fn: (tx: Transaction) => Promise): Promise; - addUnprocessedEntities( - tx: Transaction, - options: AddUnprocessedEntitiesOptions, - ): Promise; - replaceUnprocessedEntities( txOpaque: Transaction, options: ReplaceUnprocessedEntitiesOptions, diff --git a/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts b/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts index 48a40a2f6e..bb23b6afb3 100644 --- a/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts +++ b/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts @@ -68,7 +68,6 @@ export class DefaultCatalogProcessingOrchestrator async process( request: EntityProcessingRequest, ): Promise { - // TODO: implement dryRun/eager return this.processSingleEntity(request.entity); } diff --git a/plugins/catalog-backend/src/next/processing/ProcessorOutputCollector.ts b/plugins/catalog-backend/src/next/processing/ProcessorOutputCollector.ts index 3f8594bd48..507def964a 100644 --- a/plugins/catalog-backend/src/next/processing/ProcessorOutputCollector.ts +++ b/plugins/catalog-backend/src/next/processing/ProcessorOutputCollector.ts @@ -24,7 +24,12 @@ import { import { Logger } from 'winston'; import { CatalogProcessorResult } from '../../ingestion'; import { locationSpecToLocationEntity } from '../util'; -import { getEntityOriginLocationRef, validateEntityEnvelope } from './util'; +import { DeferredEntity } from './types'; +import { + getEntityLocationRef, + getEntityOriginLocationRef, + validateEntityEnvelope, +} from './util'; /** * Helper class for aggregating all of the emitted data from processors. @@ -32,7 +37,7 @@ import { getEntityOriginLocationRef, validateEntityEnvelope } from './util'; export class ProcessorOutputCollector { private readonly errors = new Array(); private readonly relations = new Array(); - private readonly deferredEntities = new Array(); + private readonly deferredEntities = new Array(); private done = false; constructor( @@ -73,6 +78,8 @@ export class ProcessorOutputCollector { return; } + const location = stringifyLocationReference(i.location); + // Note that at this point, we have only validated the envelope part of // the entity data. Annotations are not part of that, so we have to be // defensive. If the annotations were malformed (e.g. were not a valid @@ -81,7 +88,6 @@ export class ProcessorOutputCollector { const annotations = entity.metadata.annotations || {}; if (typeof annotations === 'object' && !Array.isArray(annotations)) { const originLocation = getEntityOriginLocationRef(this.parentEntity); - const location = stringifyLocationReference(i.location); entity = { ...entity, metadata: { @@ -95,11 +101,14 @@ export class ProcessorOutputCollector { }; } - this.deferredEntities.push(entity); + this.deferredEntities.push({ entity, locationKey: location }); } else if (i.type === 'location') { - this.deferredEntities.push( - locationSpecToLocationEntity(i.location, this.parentEntity), + const entity = locationSpecToLocationEntity( + i.location, + this.parentEntity, ); + const locationKey = getEntityLocationRef(entity); + this.deferredEntities.push({ entity, locationKey }); } else if (i.type === 'relation') { this.relations.push(i.relation); } else if (i.type === 'error') { diff --git a/plugins/catalog-backend/src/next/processing/types.ts b/plugins/catalog-backend/src/next/processing/types.ts index e76e789e54..9e30d07404 100644 --- a/plugins/catalog-backend/src/next/processing/types.ts +++ b/plugins/catalog-backend/src/next/processing/types.ts @@ -27,7 +27,7 @@ export type EntityProcessingResult = ok: true; state: Map; completedEntity: Entity; - deferredEntities: Entity[]; + deferredEntities: DeferredEntity[]; relations: EntityRelationSpec[]; errors: Error[]; } @@ -39,3 +39,8 @@ export type EntityProcessingResult = export interface CatalogProcessingOrchestrator { process(request: EntityProcessingRequest): Promise; } + +export type DeferredEntity = { + entity: Entity; + locationKey?: string; +}; diff --git a/plugins/catalog-backend/src/next/types.ts b/plugins/catalog-backend/src/next/types.ts index c689f288aa..3b27dd7d10 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 { DeferredEntity } from './processing/types'; export interface LocationService { createLocation( @@ -39,8 +40,8 @@ export interface CatalogProcessingEngine { } export type EntityProviderMutation = - | { type: 'full'; entities: Entity[] } - | { type: 'delta'; added: Entity[]; removed: Entity[] }; + | { type: 'full'; entities: DeferredEntity[] } + | { type: 'delta'; added: DeferredEntity[]; removed: DeferredEntity[] }; export interface EntityProviderConnection { applyMutation(mutation: EntityProviderMutation): Promise;