paulononato/backstage
1
0
Fork 0
Code Issues Pull Requests Actions 38 Packages Projects Releases Wiki Activity

Merge pull request #6266 from backstage/mob/location-key

Browse Source 
catalog-backend: introduce location key as a conflict resolution mechanism
This commit is contained in:
Johan Haals
2021-06-30 16:05:48 +02:00
committed by GitHub
parent a9697e6107 3cf22d0a92
commit 98acb3d197
17 changed files with 621 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/rude-windows-live.md
+17
View File
@@ -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.
plugins/catalog-backend/migrations/20210622104022_refresh_state_location_key.js
+40
View File
@@ -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');
});
};
plugins/catalog-backend/src/next/ConfigLocationEntityProvider.test.ts
+23 -15
View File
@@ -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',
},
]),
});
});
plugins/catalog-backend/src/next/ConfigLocationEntityProvider.ts
+4 -1
View File
@@ -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({
plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts
+5 -4
View File
@@ -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,
});
});
plugins/catalog-backend/src/next/DefaultLocationService.test.ts
+7 -4
View File
@@ -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: [],
plugins/catalog-backend/src/next/DefaultLocationService.ts
+8 -3
View File
@@ -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,
});
plugins/catalog-backend/src/next/DefaultLocationStore.test.ts
+24 -16
View File
@@ -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,
plugins/catalog-backend/src/next/DefaultLocationStore.ts
+7 -5
View File
@@ -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<DbLocationsRow>('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({
plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts
+281 -33
View File
@@ -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<DbRefreshStateRow>('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<DbRefreshStateRow>(
'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<DbRefreshStateReferencesRow>(
'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<string, JsonObject>(),
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<DbRefreshStateRow>(
'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<DbRefreshStateRow>(
'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<DbRefreshStateRow>(
'refresh_state',
).select();
expect(currentRefreshState).toEqual([
expect.objectContaining({
entity_ref: 'location:default/replaced',
location_key: 'file:///tmp/foobar',
}),
]);
const currentRefRowState = await knex<DbRefreshStateReferencesRow>(
'refresh_state_references',
).select();
expect(currentRefRowState).toEqual([
expect.objectContaining({
source_key: 'lols',
target_entity_ref: 'location:default/replaced',
}),
]);
},
60_000,
);
});
describe('getProcessableEntities', () => {
plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts
+164 -71
View File
@@ -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<DbRefreshStateRow>('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<DbRefreshStateReferencesRow>(
'refresh_state_references',
)
.where({ source_key: options.sourceKey })
.select('target_entity_ref')
.then(rows => rows.map(r => r.target_entity_ref));
.leftJoin<DbRefreshStateRow>('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<DeferredEntity>();
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<DbRefreshStateRow>[] = 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<void> {
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<DbRefreshStateRow>),
);
const stateReferenceRows = stateRows.map(
stateRow =>
({
source_entity_ref: options.entityRef,
target_entity_ref: stateRow.entity_ref,
} as Knex.DbRecord<DbRefreshStateReferencesRow>),
);
// Keeps track of the entities that we end up inserting to update refresh_state_references afterwards
const stateReferences = new Array<string>();
const conflictingStateReferences = new Array<string>();
// 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<DbRefreshStateRow>('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<DbRefreshStateRow>('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<any>({
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<DbRefreshStateRow>(
'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<DbRefreshStateReferencesRow>('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<DbRefreshStateReferencesRow>('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<DbRefreshStateReferencesRow>('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<string, JsonObject>(),
errors: i.errors,
locationKey: i.location_key,
} as RefreshStateItem),
),
};
plugins/catalog-backend/src/next/database/tables.ts
+1
View File
@@ -29,6 +29,7 @@ export type DbRefreshStateRow = {
next_update_at: string;
last_discovery_at: string; // remove?
errors?: string;
location_key?: string;
};
export type DbRefreshStateReferencesRow = {
plugins/catalog-backend/src/next/database/types.ts
+16 -13
View File
@@ -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<string, JsonObject>;
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<string, JsonObject>;
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<T>(fn: (tx: Transaction) => Promise<T>): Promise<T>;
addUnprocessedEntities(
tx: Transaction,
options: AddUnprocessedEntitiesOptions,
): Promise<void>;
replaceUnprocessedEntities(
txOpaque: Transaction,
options: ReplaceUnprocessedEntitiesOptions,
plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts
-1
View File
@@ -68,7 +68,6 @@ export class DefaultCatalogProcessingOrchestrator
async process(
request: EntityProcessingRequest,
): Promise<EntityProcessingResult> {
// TODO: implement dryRun/eager
return this.processSingleEntity(request.entity);
}
plugins/catalog-backend/src/next/processing/ProcessorOutputCollector.ts
+15 -6
View File
@@ -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<Error>();
private readonly relations = new Array<EntityRelationSpec>();
private readonly deferredEntities = new Array<Entity>();
private readonly deferredEntities = new Array<DeferredEntity>();
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') {
plugins/catalog-backend/src/next/processing/types.ts
+6 -1
View File
@@ -27,7 +27,7 @@ export type EntityProcessingResult =
ok: true;
state: Map<string, JsonObject>;
completedEntity: Entity;
deferredEntities: Entity[];
deferredEntities: DeferredEntity[];
relations: EntityRelationSpec[];
errors: Error[];
}
@@ -39,3 +39,8 @@ export type EntityProcessingResult =
export interface CatalogProcessingOrchestrator {
process(request: EntityProcessingRequest): Promise<EntityProcessingResult>;
}
export type DeferredEntity = {
entity: Entity;
locationKey?: string;
};
plugins/catalog-backend/src/next/types.ts
+3 -2
View File
@@ -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<void>;