Merge pull request #15143 from backstage/mob/catafactor
catalog-backend: split up catalog database into 3 separate implementations + operations
This commit is contained in:
Patrik Oldsberg
@@ -0,0 +1,5 @@
|
||||
---
|
||||
'@backstage/plugin-catalog-backend': patch
|
||||
---
|
||||
|
||||
Internal code reorganization.
|
||||
@@ -239,7 +239,7 @@ export type CatalogPermissionRule<
|
||||
// @alpha
|
||||
export const catalogPlugin: (options?: undefined) => BackendFeature;
|
||||
|
||||
// @public (undocumented)
|
||||
// @public
|
||||
export interface CatalogProcessingEngine {
|
||||
// (undocumented)
|
||||
start(): Promise<void>;
|
||||
|
||||
@@ -0,0 +1,110 @@
|
||||
/*
|
||||
* 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 { getVoidLogger } from '@backstage/backend-common';
|
||||
import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils';
|
||||
import { Logger } from 'winston';
|
||||
import { DefaultCatalogDatabase } from './DefaultCatalogDatabase';
|
||||
import { applyDatabaseMigrations } from './migrations';
|
||||
import { DbRefreshStateReferencesRow, DbRefreshStateRow } from './tables';
|
||||
|
||||
describe('DefaultCatalogDatabase', () => {
|
||||
const defaultLogger = getVoidLogger();
|
||||
const databases = TestDatabases.create({
|
||||
ids: ['MYSQL_8', 'POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'],
|
||||
});
|
||||
|
||||
async function createDatabase(
|
||||
databaseId: TestDatabaseId,
|
||||
logger: Logger = defaultLogger,
|
||||
) {
|
||||
const knex = await databases.init(databaseId);
|
||||
await applyDatabaseMigrations(knex);
|
||||
return {
|
||||
knex,
|
||||
db: new DefaultCatalogDatabase({
|
||||
database: knex,
|
||||
logger,
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
describe('listAncestors', () => {
|
||||
let nextId = 1;
|
||||
function makeEntity(ref: string) {
|
||||
return {
|
||||
entity_id: String(nextId++),
|
||||
entity_ref: ref,
|
||||
unprocessed_entity: JSON.stringify({
|
||||
kind: 'Location',
|
||||
apiVersion: '1.0.0',
|
||||
metadata: {
|
||||
name: 'xyz',
|
||||
},
|
||||
}),
|
||||
errors: '[]',
|
||||
next_update_at: '2019-01-01 23:00:00',
|
||||
last_discovery_at: '2021-04-01 13:37:00',
|
||||
};
|
||||
}
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should return ancestors, %p',
|
||||
async databaseId => {
|
||||
const { knex, db } = await createDatabase(databaseId);
|
||||
|
||||
await knex<DbRefreshStateRow>('refresh_state').insert(
|
||||
makeEntity('location:default/root-1'),
|
||||
);
|
||||
await knex<DbRefreshStateRow>('refresh_state').insert(
|
||||
makeEntity('location:default/root-2'),
|
||||
);
|
||||
await knex<DbRefreshStateRow>('refresh_state').insert(
|
||||
makeEntity('component:default/foobar'),
|
||||
);
|
||||
|
||||
await knex<DbRefreshStateReferencesRow>(
|
||||
'refresh_state_references',
|
||||
).insert({
|
||||
source_key: 'source',
|
||||
target_entity_ref: 'location:default/root-2',
|
||||
});
|
||||
await knex<DbRefreshStateReferencesRow>(
|
||||
'refresh_state_references',
|
||||
).insert({
|
||||
source_entity_ref: 'location:default/root-2',
|
||||
target_entity_ref: 'location:default/root-1',
|
||||
});
|
||||
await knex<DbRefreshStateReferencesRow>(
|
||||
'refresh_state_references',
|
||||
).insert({
|
||||
source_entity_ref: 'location:default/root-1',
|
||||
target_entity_ref: 'component:default/foobar',
|
||||
});
|
||||
|
||||
const result = await db.transaction(tx =>
|
||||
db.listAncestors(tx, {
|
||||
entityRef: 'component:default/foobar',
|
||||
}),
|
||||
);
|
||||
expect(result.entityRefs).toEqual([
|
||||
'location:default/root-1',
|
||||
'location:default/root-2',
|
||||
]);
|
||||
},
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,113 @@
|
||||
/*
|
||||
* 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.
|
||||
*/
|
||||
|
||||
import { NotFoundError } from '@backstage/errors';
|
||||
import { Knex } from 'knex';
|
||||
import type { Logger } from 'winston';
|
||||
import {
|
||||
CatalogDatabase,
|
||||
ListAncestorsOptions,
|
||||
ListAncestorsResult,
|
||||
RefreshOptions,
|
||||
} from './types';
|
||||
import { DbRefreshStateReferencesRow, DbRefreshStateRow } from './tables';
|
||||
import { rethrowError } from './conversion';
|
||||
import { Transaction } from './types';
|
||||
|
||||
const MAX_ANCESTOR_DEPTH = 32;
|
||||
|
||||
export class DefaultCatalogDatabase implements CatalogDatabase {
|
||||
constructor(
|
||||
private readonly options: {
|
||||
database: Knex;
|
||||
logger: Logger;
|
||||
},
|
||||
) {}
|
||||
|
||||
async transaction<T>(fn: (tx: Transaction) => Promise<T>): Promise<T> {
|
||||
try {
|
||||
let result: T | undefined = undefined;
|
||||
|
||||
await this.options.database.transaction(
|
||||
async tx => {
|
||||
// We can't return here, as knex swallows the return type in case the transaction is rolled back:
|
||||
// https://github.com/knex/knex/blob/e37aeaa31c8ef9c1b07d2e4d3ec6607e557d800d/lib/transaction.js#L136
|
||||
result = await fn(tx);
|
||||
},
|
||||
{
|
||||
// If we explicitly trigger a rollback, don't fail.
|
||||
doNotRejectOnRollback: true,
|
||||
},
|
||||
);
|
||||
|
||||
return result!;
|
||||
} catch (e) {
|
||||
this.options.logger.debug(`Error during transaction, ${e}`);
|
||||
throw rethrowError(e);
|
||||
}
|
||||
}
|
||||
|
||||
async listAncestors(
|
||||
txOpaque: Transaction,
|
||||
options: ListAncestorsOptions,
|
||||
): Promise<ListAncestorsResult> {
|
||||
const tx = txOpaque as Knex.Transaction;
|
||||
const { entityRef } = options;
|
||||
const entityRefs = new Array<string>();
|
||||
|
||||
let currentRef = entityRef.toLocaleLowerCase('en-US');
|
||||
for (let depth = 1; depth <= MAX_ANCESTOR_DEPTH; depth += 1) {
|
||||
const rows = await tx<DbRefreshStateReferencesRow>(
|
||||
'refresh_state_references',
|
||||
)
|
||||
.where({ target_entity_ref: currentRef })
|
||||
.select();
|
||||
|
||||
if (rows.length === 0) {
|
||||
if (depth === 1) {
|
||||
throw new NotFoundError(`Entity ${currentRef} not found`);
|
||||
}
|
||||
throw new NotFoundError(
|
||||
`Entity ${entityRef} has a broken parent reference chain at ${currentRef}`,
|
||||
);
|
||||
}
|
||||
|
||||
const parentRef = rows.find(r => r.source_entity_ref)?.source_entity_ref;
|
||||
if (!parentRef) {
|
||||
// We've reached the top of the tree which is the entityProvider.
|
||||
// In this case we refresh the entity itself.
|
||||
return { entityRefs };
|
||||
}
|
||||
entityRefs.push(parentRef);
|
||||
currentRef = parentRef;
|
||||
}
|
||||
throw new Error(
|
||||
`Unable receive ancestors for ${entityRef}, reached maximum depth of ${MAX_ANCESTOR_DEPTH}`,
|
||||
);
|
||||
}
|
||||
|
||||
async refresh(txOpaque: Transaction, options: RefreshOptions): Promise<void> {
|
||||
const tx = txOpaque as Knex.Transaction;
|
||||
const { entityRef } = options;
|
||||
|
||||
const updateResult = await tx<DbRefreshStateRow>('refresh_state')
|
||||
.where({ entity_ref: entityRef.toLocaleLowerCase('en-US') })
|
||||
.update({ next_update_at: tx.fn.now() });
|
||||
if (updateResult === 0) {
|
||||
throw new NotFoundError(`Failed to schedule ${entityRef} for refresh`);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -33,7 +33,7 @@ import { createRandomProcessingInterval } from '../processing/refresh';
|
||||
import { timestampToDateTime } from './conversion';
|
||||
import { generateStableHash } from './util';
|
||||
|
||||
describe('Default Processing Database', () => {
|
||||
describe('DefaultProcessingDatabase', () => {
|
||||
const defaultLogger = getVoidLogger();
|
||||
const databases = TestDatabases.create({
|
||||
ids: ['MYSQL_8', 'POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'],
|
||||
@@ -582,664 +582,6 @@ describe('Default Processing Database', () => {
|
||||
);
|
||||
});
|
||||
|
||||
describe('replaceUnprocessedEntities', () => {
|
||||
const createLocations = async (db: Knex, entityRefs: string[]) => {
|
||||
for (const ref of entityRefs) {
|
||||
await insertRefreshStateRow(db, {
|
||||
entity_id: uuid.v4(),
|
||||
entity_ref: ref,
|
||||
unprocessed_entity: '{}',
|
||||
processed_entity: '{}',
|
||||
errors: '[]',
|
||||
next_update_at: '2021-04-01 13:37:00',
|
||||
last_discovery_at: '2021-04-01 13:37:00',
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'replaces all existing state correctly for simple dependency chains, %p',
|
||||
async databaseId => {
|
||||
const { knex, db } = await createDatabase(databaseId);
|
||||
/*
|
||||
config -> location:default/root -> location:default/root-1 -> location:default/root-2
|
||||
database -> location:default/second -> location:default/root-2
|
||||
*/
|
||||
await createLocations(knex, [
|
||||
'location:default/root',
|
||||
'location:default/root-1',
|
||||
'location:default/root-2',
|
||||
'location:default/second',
|
||||
]);
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_key: 'config',
|
||||
target_entity_ref: 'location:default/root',
|
||||
});
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_key: 'database',
|
||||
target_entity_ref: 'location:default/second',
|
||||
});
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_entity_ref: 'location:default/root',
|
||||
target_entity_ref: 'location:default/root-1',
|
||||
});
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_entity_ref: 'location:default/root-1',
|
||||
target_entity_ref: 'location:default/root-2',
|
||||
});
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_entity_ref: 'location:default/second',
|
||||
target_entity_ref: 'location:default/root-2',
|
||||
});
|
||||
|
||||
await db.transaction(tx =>
|
||||
db.replaceUnprocessedEntities(tx, {
|
||||
type: 'full',
|
||||
sourceKey: 'config',
|
||||
items: [
|
||||
{
|
||||
entity: {
|
||||
apiVersion: '1.0.0',
|
||||
metadata: {
|
||||
name: 'new-root',
|
||||
},
|
||||
kind: 'Location',
|
||||
} as Entity,
|
||||
locationKey: 'file:///tmp/foobar',
|
||||
},
|
||||
],
|
||||
}),
|
||||
);
|
||||
|
||||
const currentRefreshState = await knex<DbRefreshStateRow>(
|
||||
'refresh_state',
|
||||
).select();
|
||||
|
||||
const currentRefRowState = await knex<DbRefreshStateReferencesRow>(
|
||||
'refresh_state_references',
|
||||
).select();
|
||||
|
||||
for (const ref of [
|
||||
'location:default/root',
|
||||
'location:default/root-1',
|
||||
]) {
|
||||
expect(
|
||||
currentRefreshState.some(t => t.entity_ref === ref),
|
||||
).toBeFalsy();
|
||||
}
|
||||
|
||||
expect(
|
||||
currentRefreshState.some(
|
||||
t => t.entity_ref === 'location:default/new-root',
|
||||
),
|
||||
).toBeTruthy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.source_entity_ref === 'location:default/root' &&
|
||||
t.target_entity_ref === 'location:default/root-1',
|
||||
),
|
||||
).toBeFalsy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.source_entity_ref === 'location:default/root-1' &&
|
||||
t.target_entity_ref === 'location:default/root-2',
|
||||
),
|
||||
).toBeFalsy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.target_entity_ref === 'location:default/root-1' &&
|
||||
t.source_key === 'config',
|
||||
),
|
||||
).toBeFalsy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.target_entity_ref === 'location:default/new-root' &&
|
||||
t.source_key === 'config',
|
||||
),
|
||||
).toBeTruthy();
|
||||
},
|
||||
60_000,
|
||||
);
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should work for more complex chains, %p',
|
||||
async databaseId => {
|
||||
const { knex, db } = await createDatabase(databaseId);
|
||||
/*
|
||||
config -> location:default/root -> location:default/root-1 -> location:default/root-2
|
||||
config -> location:default/root -> location:default/root-1a -> location:default/root-2
|
||||
*/
|
||||
await createLocations(knex, [
|
||||
'location:default/root',
|
||||
'location:default/root-1',
|
||||
'location:default/root-2',
|
||||
'location:default/root-1a',
|
||||
]);
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_key: 'config',
|
||||
target_entity_ref: 'location:default/root',
|
||||
});
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_entity_ref: 'location:default/root',
|
||||
target_entity_ref: 'location:default/root-1',
|
||||
});
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_entity_ref: 'location:default/root',
|
||||
target_entity_ref: 'location:default/root-1a',
|
||||
});
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_entity_ref: 'location:default/root-1',
|
||||
target_entity_ref: 'location:default/root-2',
|
||||
});
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_entity_ref: 'location:default/root-1a',
|
||||
target_entity_ref: 'location:default/root-2',
|
||||
});
|
||||
|
||||
await db.transaction(async tx => {
|
||||
await db.replaceUnprocessedEntities(tx, {
|
||||
type: 'full',
|
||||
sourceKey: 'config',
|
||||
items: [
|
||||
{
|
||||
entity: {
|
||||
apiVersion: '1.0.0',
|
||||
metadata: {
|
||||
name: 'new-root',
|
||||
},
|
||||
kind: 'Location',
|
||||
} as Entity,
|
||||
locationKey: 'file:/tmp/foobar',
|
||||
},
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
const currentRefreshState = await knex<DbRefreshStateRow>(
|
||||
'refresh_state',
|
||||
).select();
|
||||
|
||||
const currentRefRowState = await knex<DbRefreshStateReferencesRow>(
|
||||
'refresh_state_references',
|
||||
).select();
|
||||
|
||||
const deletedRefs = [
|
||||
'location:default/root',
|
||||
'location:default/root-1',
|
||||
'location:default/root-1a',
|
||||
'location:default/root-2',
|
||||
];
|
||||
|
||||
for (const ref of deletedRefs) {
|
||||
expect(
|
||||
currentRefreshState.some(t => t.entity_ref === ref),
|
||||
).toBeFalsy();
|
||||
}
|
||||
|
||||
expect(
|
||||
currentRefreshState.some(
|
||||
t => t.entity_ref === 'location:default/new-root',
|
||||
),
|
||||
).toBeTruthy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.source_key === 'config' &&
|
||||
t.target_entity_ref === 'location:default/new-root',
|
||||
),
|
||||
).toBeTruthy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.source_key === 'config' &&
|
||||
t.target_entity_ref === 'location:default/root',
|
||||
),
|
||||
).toBeFalsy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.source_entity_ref === 'location:default/root' &&
|
||||
t.target_entity_ref === 'location:default/root-1',
|
||||
),
|
||||
).toBeFalsy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.source_entity_ref === 'location:default/root' &&
|
||||
t.target_entity_ref === 'location:default/root-1a',
|
||||
),
|
||||
).toBeFalsy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.source_entity_ref === 'location:default/root-1' &&
|
||||
t.target_entity_ref === 'location:default/root-2',
|
||||
),
|
||||
).toBeFalsy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.source_entity_ref === 'location:default/root-1a' &&
|
||||
t.target_entity_ref === 'location:default/root-2',
|
||||
),
|
||||
).toBeFalsy();
|
||||
},
|
||||
60_000,
|
||||
);
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should add new locations using the delta options, %p',
|
||||
async databaseId => {
|
||||
const { knex, db } = await createDatabase(databaseId);
|
||||
|
||||
// Existing state and references should stay
|
||||
await createLocations(knex, ['location:default/existing']);
|
||||
await insertRefRow(knex, {
|
||||
source_key: 'lols',
|
||||
target_entity_ref: 'location:default/existing',
|
||||
});
|
||||
|
||||
await db.transaction(async tx => {
|
||||
await db.replaceUnprocessedEntities(tx, {
|
||||
type: 'delta',
|
||||
sourceKey: 'lols',
|
||||
removed: [],
|
||||
added: [
|
||||
{
|
||||
entity: {
|
||||
apiVersion: '1.0.0',
|
||||
metadata: {
|
||||
name: 'new-root',
|
||||
},
|
||||
kind: 'Location',
|
||||
} as Entity,
|
||||
locationKey: 'file:///tmp/foobar',
|
||||
},
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
const currentRefreshState = await knex<DbRefreshStateRow>(
|
||||
'refresh_state',
|
||||
).select();
|
||||
|
||||
const currentRefRowState = await knex<DbRefreshStateReferencesRow>(
|
||||
'refresh_state_references',
|
||||
).select();
|
||||
|
||||
expect(
|
||||
currentRefreshState.some(
|
||||
t => t.entity_ref === 'location:default/new-root',
|
||||
),
|
||||
).toBeTruthy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.source_key === 'lols' &&
|
||||
t.target_entity_ref === 'location:default/new-root',
|
||||
),
|
||||
).toBeTruthy();
|
||||
|
||||
expect(
|
||||
currentRefreshState.some(
|
||||
t => t.entity_ref === 'location:default/existing',
|
||||
),
|
||||
).toBeTruthy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.source_key === 'lols' &&
|
||||
t.target_entity_ref === 'location:default/existing',
|
||||
),
|
||||
).toBeTruthy();
|
||||
},
|
||||
60_000,
|
||||
);
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should not remove locations that are referenced elsewhere, %p',
|
||||
async databaseId => {
|
||||
const { knex, db } = await createDatabase(databaseId);
|
||||
/*
|
||||
config-1 -> location:default/root
|
||||
config-2 -> location:default/root
|
||||
*/
|
||||
await createLocations(knex, ['location:default/root']);
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_key: 'config-1',
|
||||
target_entity_ref: 'location:default/root',
|
||||
});
|
||||
await insertRefRow(knex, {
|
||||
source_key: 'config-2',
|
||||
target_entity_ref: 'location:default/root',
|
||||
});
|
||||
|
||||
await db.transaction(async tx => {
|
||||
await db.replaceUnprocessedEntities(tx, {
|
||||
type: 'full',
|
||||
sourceKey: 'config-1',
|
||||
items: [],
|
||||
});
|
||||
});
|
||||
|
||||
const currentRefreshState = await knex<DbRefreshStateRow>(
|
||||
'refresh_state',
|
||||
).select();
|
||||
|
||||
const currentRefRowState = await knex<DbRefreshStateReferencesRow>(
|
||||
'refresh_state_references',
|
||||
).select();
|
||||
|
||||
expect(currentRefRowState).toEqual([
|
||||
expect.objectContaining({
|
||||
source_key: 'config-2',
|
||||
target_entity_ref: 'location:default/root',
|
||||
}),
|
||||
]);
|
||||
|
||||
expect(currentRefreshState).toEqual([
|
||||
expect.objectContaining({
|
||||
entity_ref: 'location:default/root',
|
||||
}),
|
||||
]);
|
||||
},
|
||||
60_000,
|
||||
);
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should remove old locations using the delta options, %p',
|
||||
async databaseId => {
|
||||
const { knex, db } = await createDatabase(databaseId);
|
||||
await createLocations(knex, ['location:default/new-root']);
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_key: 'lols',
|
||||
target_entity_ref: 'location:default/new-root',
|
||||
});
|
||||
|
||||
await db.transaction(async tx => {
|
||||
await db.replaceUnprocessedEntities(tx, {
|
||||
type: 'delta',
|
||||
sourceKey: 'lols',
|
||||
added: [],
|
||||
removed: [
|
||||
{
|
||||
entityRef: 'location:default/new-root',
|
||||
locationKey: 'file:/tmp/foobar',
|
||||
},
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
const currentRefreshState = await knex<DbRefreshStateRow>(
|
||||
'refresh_state',
|
||||
).select();
|
||||
|
||||
const currentRefRowState = await knex<DbRefreshStateReferencesRow>(
|
||||
'refresh_state_references',
|
||||
).select();
|
||||
|
||||
expect(
|
||||
currentRefreshState.some(
|
||||
t => t.entity_ref === 'location:default/new-root',
|
||||
),
|
||||
).toBeFalsy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.source_key === 'lols' &&
|
||||
t.target_entity_ref === 'location:default/new-root',
|
||||
),
|
||||
).toBeFalsy();
|
||||
},
|
||||
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,
|
||||
);
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should support replacing modified entities during a full update, %p',
|
||||
async databaseId => {
|
||||
const { knex, db } = await createDatabase(databaseId);
|
||||
|
||||
await db.transaction(async tx => {
|
||||
await db.replaceUnprocessedEntities(tx, {
|
||||
type: 'full',
|
||||
sourceKey: 'lols',
|
||||
items: [
|
||||
{
|
||||
entity: {
|
||||
apiVersion: '1',
|
||||
kind: 'Component',
|
||||
metadata: { name: 'a' },
|
||||
spec: { marker: 'WILL_CHANGE' },
|
||||
} as Entity,
|
||||
locationKey: 'file:///tmp/a',
|
||||
},
|
||||
{
|
||||
entity: {
|
||||
apiVersion: '1',
|
||||
kind: 'Component',
|
||||
metadata: { name: 'b' },
|
||||
spec: { marker: 'NEVER_CHANGES' },
|
||||
} as Entity,
|
||||
locationKey: 'file:///tmp/b',
|
||||
},
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
let state = await knex<DbRefreshStateRow>('refresh_state').select();
|
||||
expect(state).toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
entity_ref: 'component:default/a',
|
||||
location_key: 'file:///tmp/a',
|
||||
unprocessed_entity: expect.stringContaining('WILL_CHANGE'),
|
||||
}),
|
||||
expect.objectContaining({
|
||||
entity_ref: 'component:default/b',
|
||||
location_key: 'file:///tmp/b',
|
||||
unprocessed_entity: expect.stringContaining('NEVER_CHANGES'),
|
||||
}),
|
||||
]),
|
||||
);
|
||||
|
||||
await db.transaction(async tx => {
|
||||
await db.replaceUnprocessedEntities(tx, {
|
||||
type: 'full',
|
||||
sourceKey: 'lols',
|
||||
items: [
|
||||
{
|
||||
entity: {
|
||||
apiVersion: '1',
|
||||
kind: 'Component',
|
||||
metadata: { name: 'a' },
|
||||
spec: { marker: 'HAS_CHANGED' },
|
||||
} as Entity,
|
||||
locationKey: 'file:///tmp/a',
|
||||
},
|
||||
{
|
||||
entity: {
|
||||
apiVersion: '1',
|
||||
kind: 'Component',
|
||||
metadata: { name: 'b' },
|
||||
spec: { marker: 'NEVER_CHANGES' },
|
||||
} as Entity,
|
||||
locationKey: 'file:///tmp/b',
|
||||
},
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
state = await knex<DbRefreshStateRow>('refresh_state').select();
|
||||
expect(state).toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
entity_ref: 'component:default/a',
|
||||
location_key: 'file:///tmp/a',
|
||||
unprocessed_entity: expect.stringContaining('HAS_CHANGED'),
|
||||
}),
|
||||
expect.objectContaining({
|
||||
entity_ref: 'component:default/b',
|
||||
location_key: 'file:///tmp/b',
|
||||
unprocessed_entity: expect.stringContaining('NEVER_CHANGES'),
|
||||
}),
|
||||
]),
|
||||
);
|
||||
},
|
||||
60_000,
|
||||
);
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should successfully fall back from batch to individual mode on conflicts, %p',
|
||||
async databaseId => {
|
||||
const fakeLogger = {
|
||||
debug: jest.fn(),
|
||||
};
|
||||
const { knex, db } = await createDatabase(
|
||||
databaseId,
|
||||
fakeLogger as any,
|
||||
);
|
||||
|
||||
await createLocations(knex, ['component:default/a']);
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_key: undefined,
|
||||
target_entity_ref: 'component:default/a',
|
||||
});
|
||||
|
||||
await db.transaction(async tx => {
|
||||
await db.replaceUnprocessedEntities(tx, {
|
||||
type: 'full',
|
||||
sourceKey: 'lols',
|
||||
items: [
|
||||
{
|
||||
entity: {
|
||||
apiVersion: '1',
|
||||
kind: 'Component',
|
||||
metadata: { name: 'a' },
|
||||
spec: { marker: 'WILL_CHANGE' },
|
||||
} as Entity,
|
||||
locationKey: 'file:///tmp/a',
|
||||
},
|
||||
],
|
||||
});
|
||||
});
|
||||
expect(fakeLogger.debug).toHaveBeenCalledWith(
|
||||
expect.stringMatching(
|
||||
/Fast insert path failed, falling back to slow path/,
|
||||
),
|
||||
);
|
||||
|
||||
const state = await knex<DbRefreshStateRow>('refresh_state').select();
|
||||
expect(state).toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
entity_ref: 'component:default/a',
|
||||
location_key: 'file:///tmp/a',
|
||||
unprocessed_entity: expect.stringContaining('WILL_CHANGE'),
|
||||
}),
|
||||
]),
|
||||
);
|
||||
},
|
||||
60_000,
|
||||
);
|
||||
});
|
||||
|
||||
describe('getProcessableEntities', () => {
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should return entities to process, %p',
|
||||
@@ -1329,64 +671,6 @@ describe('Default Processing Database', () => {
|
||||
);
|
||||
});
|
||||
|
||||
describe('listAncestors', () => {
|
||||
let nextId = 1;
|
||||
function makeEntity(ref: string) {
|
||||
return {
|
||||
entity_id: String(nextId++),
|
||||
entity_ref: ref,
|
||||
unprocessed_entity: JSON.stringify({
|
||||
kind: 'Location',
|
||||
apiVersion: '1.0.0',
|
||||
metadata: {
|
||||
name: 'xyz',
|
||||
},
|
||||
}),
|
||||
errors: '[]',
|
||||
next_update_at: '2019-01-01 23:00:00',
|
||||
last_discovery_at: '2021-04-01 13:37:00',
|
||||
};
|
||||
}
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should return ancestors, %p',
|
||||
async databaseId => {
|
||||
const { knex, db } = await createDatabase(databaseId);
|
||||
|
||||
await knex<DbRefreshStateRow>('refresh_state').insert(
|
||||
makeEntity('location:default/root-1'),
|
||||
);
|
||||
await knex<DbRefreshStateRow>('refresh_state').insert(
|
||||
makeEntity('location:default/root-2'),
|
||||
);
|
||||
await knex<DbRefreshStateRow>('refresh_state').insert(
|
||||
makeEntity('component:default/foobar'),
|
||||
);
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_key: 'source',
|
||||
target_entity_ref: 'location:default/root-2',
|
||||
});
|
||||
await insertRefRow(knex, {
|
||||
source_entity_ref: 'location:default/root-2',
|
||||
target_entity_ref: 'location:default/root-1',
|
||||
});
|
||||
await insertRefRow(knex, {
|
||||
source_entity_ref: 'location:default/root-1',
|
||||
target_entity_ref: 'component:default/foobar',
|
||||
});
|
||||
|
||||
const result = await db.transaction(async tx =>
|
||||
db.listAncestors(tx, { entityRef: 'component:default/foobar' }),
|
||||
);
|
||||
expect(result.entityRefs).toEqual([
|
||||
'location:default/root-1',
|
||||
'location:default/root-2',
|
||||
]);
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
describe('listParents', () => {
|
||||
let nextId = 1;
|
||||
function makeEntity(ref: string) {
|
||||
|
||||
@@ -15,26 +15,10 @@
|
||||
*/
|
||||
|
||||
import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
|
||||
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,
|
||||
GetProcessableEntitiesResult,
|
||||
ProcessingDatabase,
|
||||
RefreshStateItem,
|
||||
RefreshOptions,
|
||||
ReplaceUnprocessedEntitiesOptions,
|
||||
UpdateProcessedEntityOptions,
|
||||
ListAncestorsOptions,
|
||||
ListAncestorsResult,
|
||||
UpdateEntityCacheOptions,
|
||||
ListParentsOptions,
|
||||
ListParentsResult,
|
||||
RefreshByKeyOptions,
|
||||
} from './types';
|
||||
import { ProcessingIntervalFunction } from '../processing/refresh';
|
||||
import { rethrowError, timestampToDateTime } from './conversion';
|
||||
import { initDatabaseMetrics } from './metrics';
|
||||
@@ -44,17 +28,28 @@ import {
|
||||
DbRefreshStateRow,
|
||||
DbRelationsRow,
|
||||
} from './tables';
|
||||
import {
|
||||
GetProcessableEntitiesResult,
|
||||
ListParentsOptions,
|
||||
ListParentsResult,
|
||||
ProcessingDatabase,
|
||||
RefreshStateItem,
|
||||
Transaction,
|
||||
UpdateEntityCacheOptions,
|
||||
UpdateProcessedEntityOptions,
|
||||
} from './types';
|
||||
|
||||
import { generateStableHash } from './util';
|
||||
import { isDatabaseConflictError } from '@backstage/backend-common';
|
||||
import { DeferredEntity } from '@backstage/plugin-catalog-node';
|
||||
import { checkLocationKeyConflict } from './operations/refreshState/checkLocationKeyConflict';
|
||||
import { insertUnprocessedEntity } from './operations/refreshState/insertUnprocessedEntity';
|
||||
import { updateUnprocessedEntity } from './operations/refreshState/updateUnprocessedEntity';
|
||||
import { generateStableHash } from './util';
|
||||
|
||||
// The number of items that are sent per batch to the database layer, when
|
||||
// doing .batchInsert calls to knex. This needs to be low enough to not cause
|
||||
// errors in the underlying engine due to exceeding query limits, but large
|
||||
// enough to get the speed benefits.
|
||||
const BATCH_SIZE = 50;
|
||||
const MAX_ANCESTOR_DEPTH = 32;
|
||||
|
||||
export class DefaultProcessingDatabase implements ProcessingDatabase {
|
||||
constructor(
|
||||
@@ -193,236 +188,6 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
|
||||
.where('entity_id', id);
|
||||
}
|
||||
|
||||
async replaceUnprocessedEntities(
|
||||
txOpaque: Transaction,
|
||||
options: ReplaceUnprocessedEntitiesOptions,
|
||||
): Promise<void> {
|
||||
const tx = txOpaque as Knex.Transaction;
|
||||
|
||||
const { toAdd, toUpsert, toRemove } = await this.createDelta(tx, options);
|
||||
|
||||
if (toRemove.length) {
|
||||
let removedCount = 0;
|
||||
const rootId = () => {
|
||||
if (tx.client.config.client.includes('mysql')) {
|
||||
return tx.raw('CAST(NULL as UNSIGNED INT)', []);
|
||||
}
|
||||
|
||||
return tx.raw('CAST(NULL as INT)', []);
|
||||
};
|
||||
for (const refs of lodash.chunk(toRemove, 1000)) {
|
||||
/*
|
||||
WITH RECURSIVE
|
||||
-- All the nodes that can be reached downwards from our root
|
||||
descendants(root_id, entity_ref) AS (
|
||||
SELECT id, target_entity_ref
|
||||
FROM refresh_state_references
|
||||
WHERE source_key = "R1" AND target_entity_ref = "A"
|
||||
UNION
|
||||
SELECT descendants.root_id, target_entity_ref
|
||||
FROM descendants
|
||||
JOIN refresh_state_references ON source_entity_ref = descendants.entity_ref
|
||||
),
|
||||
-- All the nodes that can be reached upwards from the descendants
|
||||
ancestors(root_id, via_entity_ref, to_entity_ref) AS (
|
||||
SELECT CAST(NULL as INT), entity_ref, entity_ref
|
||||
FROM descendants
|
||||
UNION
|
||||
SELECT
|
||||
CASE WHEN source_key IS NOT NULL THEN id ELSE NULL END,
|
||||
source_entity_ref,
|
||||
ancestors.to_entity_ref
|
||||
FROM ancestors
|
||||
JOIN refresh_state_references ON target_entity_ref = ancestors.via_entity_ref
|
||||
)
|
||||
-- Start out with all of the descendants
|
||||
SELECT descendants.entity_ref
|
||||
FROM descendants
|
||||
-- Expand with all ancestors that point to those, but aren't the current root
|
||||
LEFT OUTER JOIN ancestors
|
||||
ON ancestors.to_entity_ref = descendants.entity_ref
|
||||
AND ancestors.root_id IS NOT NULL
|
||||
AND ancestors.root_id != descendants.root_id
|
||||
-- Exclude all lines that had such a foreign ancestor
|
||||
WHERE ancestors.root_id IS NULL;
|
||||
*/
|
||||
removedCount += await tx<DbRefreshStateRow>('refresh_state')
|
||||
.whereIn('entity_ref', function orphanedEntityRefs(orphans) {
|
||||
return (
|
||||
orphans
|
||||
// All the nodes that can be reached downwards from our root
|
||||
.withRecursive('descendants', function descendants(outer) {
|
||||
return outer
|
||||
.select({ root_id: 'id', entity_ref: 'target_entity_ref' })
|
||||
.from('refresh_state_references')
|
||||
.where('source_key', options.sourceKey)
|
||||
.whereIn('target_entity_ref', refs)
|
||||
.union(function recursive(inner) {
|
||||
return inner
|
||||
.select({
|
||||
root_id: 'descendants.root_id',
|
||||
entity_ref:
|
||||
'refresh_state_references.target_entity_ref',
|
||||
})
|
||||
.from('descendants')
|
||||
.join('refresh_state_references', {
|
||||
'descendants.entity_ref':
|
||||
'refresh_state_references.source_entity_ref',
|
||||
});
|
||||
});
|
||||
})
|
||||
// All the nodes that can be reached upwards from the descendants
|
||||
.withRecursive('ancestors', function ancestors(outer) {
|
||||
return outer
|
||||
.select({
|
||||
root_id: rootId(),
|
||||
via_entity_ref: 'entity_ref',
|
||||
to_entity_ref: 'entity_ref',
|
||||
})
|
||||
.from('descendants')
|
||||
.union(function recursive(inner) {
|
||||
return inner
|
||||
.select({
|
||||
root_id: tx.raw(
|
||||
'CASE WHEN source_key IS NOT NULL THEN id ELSE NULL END',
|
||||
[],
|
||||
),
|
||||
via_entity_ref: 'source_entity_ref',
|
||||
to_entity_ref: 'ancestors.to_entity_ref',
|
||||
})
|
||||
.from('ancestors')
|
||||
.join('refresh_state_references', {
|
||||
target_entity_ref: 'ancestors.via_entity_ref',
|
||||
});
|
||||
});
|
||||
})
|
||||
// Start out with all of the descendants
|
||||
.select('descendants.entity_ref')
|
||||
.from('descendants')
|
||||
// Expand with all ancestors that point to those, but aren't the current root
|
||||
.leftOuterJoin('ancestors', function keepaliveRoots() {
|
||||
this.on(
|
||||
'ancestors.to_entity_ref',
|
||||
'=',
|
||||
'descendants.entity_ref',
|
||||
);
|
||||
this.andOnNotNull('ancestors.root_id');
|
||||
this.andOn('ancestors.root_id', '!=', 'descendants.root_id');
|
||||
})
|
||||
.whereNull('ancestors.root_id')
|
||||
);
|
||||
})
|
||||
.delete();
|
||||
|
||||
await tx<DbRefreshStateReferencesRow>('refresh_state_references')
|
||||
.where('source_key', '=', options.sourceKey)
|
||||
.whereIn('target_entity_ref', refs)
|
||||
.delete();
|
||||
}
|
||||
|
||||
this.options.logger.debug(
|
||||
`removed, ${removedCount} entities: ${JSON.stringify(toRemove)}`,
|
||||
);
|
||||
}
|
||||
|
||||
if (toAdd.length) {
|
||||
// The reason for this chunking, rather than just massively batch
|
||||
// inserting the entire payload, is that we fall back to the individual
|
||||
// upsert mechanism below on conflicts. That path is massively slower than
|
||||
// the fast batch path, so we don't want to end up accidentally having to
|
||||
// for example item-by-item upsert tens of thousands of entities in a
|
||||
// large initial delivery dump. The implication is that the size of these
|
||||
// chunks needs to weigh the benefit of fast successful inserts, against
|
||||
// the drawback of super slow but more rare fallbacks. There's quickly
|
||||
// diminishing returns though with turning up this value way high.
|
||||
for (const chunk of lodash.chunk(toAdd, 50)) {
|
||||
try {
|
||||
await tx.batchInsert(
|
||||
'refresh_state',
|
||||
chunk.map(item => ({
|
||||
entity_id: uuid(),
|
||||
entity_ref: stringifyEntityRef(item.deferred.entity),
|
||||
unprocessed_entity: JSON.stringify(item.deferred.entity),
|
||||
unprocessed_hash: item.hash,
|
||||
errors: '',
|
||||
location_key: item.deferred.locationKey,
|
||||
next_update_at: tx.fn.now(),
|
||||
last_discovery_at: tx.fn.now(),
|
||||
})),
|
||||
BATCH_SIZE,
|
||||
);
|
||||
await tx.batchInsert(
|
||||
'refresh_state_references',
|
||||
chunk.map(item => ({
|
||||
source_key: options.sourceKey,
|
||||
target_entity_ref: stringifyEntityRef(item.deferred.entity),
|
||||
})),
|
||||
BATCH_SIZE,
|
||||
);
|
||||
} catch (error) {
|
||||
if (!isDatabaseConflictError(error)) {
|
||||
throw error;
|
||||
} else {
|
||||
this.options.logger.debug(
|
||||
`Fast insert path failed, falling back to slow path, ${error}`,
|
||||
);
|
||||
toUpsert.push(...chunk);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (toUpsert.length) {
|
||||
for (const {
|
||||
deferred: { entity, locationKey },
|
||||
hash,
|
||||
} of toUpsert) {
|
||||
const entityRef = stringifyEntityRef(entity);
|
||||
|
||||
try {
|
||||
let ok = await this.updateUnprocessedEntity(
|
||||
tx,
|
||||
entity,
|
||||
hash,
|
||||
locationKey,
|
||||
);
|
||||
if (!ok) {
|
||||
ok = await this.insertUnprocessedEntity(
|
||||
tx,
|
||||
entity,
|
||||
hash,
|
||||
locationKey,
|
||||
);
|
||||
}
|
||||
|
||||
if (ok) {
|
||||
await tx<DbRefreshStateReferencesRow>(
|
||||
'refresh_state_references',
|
||||
).insert({
|
||||
source_key: options.sourceKey,
|
||||
target_entity_ref: entityRef,
|
||||
});
|
||||
} else {
|
||||
const conflictingKey = await this.checkLocationKeyConflict(
|
||||
tx,
|
||||
entityRef,
|
||||
locationKey,
|
||||
);
|
||||
if (conflictingKey) {
|
||||
this.options.logger.warn(
|
||||
`Source ${options.sourceKey} detected conflicting entityRef ${entityRef} already referenced by ${conflictingKey} and now also ${locationKey}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
this.options.logger.error(
|
||||
`Failed to add '${entityRef}' from source '${options.sourceKey}', ${error}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async getProcessableEntities(
|
||||
txOpaque: Transaction,
|
||||
request: { processBatchSize: number },
|
||||
@@ -487,45 +252,6 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
|
||||
};
|
||||
}
|
||||
|
||||
async listAncestors(
|
||||
txOpaque: Transaction,
|
||||
options: ListAncestorsOptions,
|
||||
): Promise<ListAncestorsResult> {
|
||||
const tx = txOpaque as Knex.Transaction;
|
||||
const { entityRef } = options;
|
||||
const entityRefs = new Array<string>();
|
||||
|
||||
let currentRef = entityRef.toLocaleLowerCase('en-US');
|
||||
for (let depth = 1; depth <= MAX_ANCESTOR_DEPTH; depth += 1) {
|
||||
const rows = await tx<DbRefreshStateReferencesRow>(
|
||||
'refresh_state_references',
|
||||
)
|
||||
.where({ target_entity_ref: currentRef })
|
||||
.select();
|
||||
|
||||
if (rows.length === 0) {
|
||||
if (depth === 1) {
|
||||
throw new NotFoundError(`Entity ${currentRef} not found`);
|
||||
}
|
||||
throw new NotFoundError(
|
||||
`Entity ${entityRef} has a broken parent reference chain at ${currentRef}`,
|
||||
);
|
||||
}
|
||||
|
||||
const parentRef = rows.find(r => r.source_entity_ref)?.source_entity_ref;
|
||||
if (!parentRef) {
|
||||
// We've reached the top of the tree which is the entityProvider.
|
||||
// In this case we refresh the entity itself.
|
||||
return { entityRefs };
|
||||
}
|
||||
entityRefs.push(parentRef);
|
||||
currentRef = parentRef;
|
||||
}
|
||||
throw new Error(
|
||||
`Unable receive ancestors for ${entityRef}, reached maximum depth of ${MAX_ANCESTOR_DEPTH}`,
|
||||
);
|
||||
}
|
||||
|
||||
async listParents(
|
||||
txOpaque: Transaction,
|
||||
options: ListParentsOptions,
|
||||
@@ -543,37 +269,6 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
|
||||
return { entityRefs };
|
||||
}
|
||||
|
||||
async refresh(txOpaque: Transaction, options: RefreshOptions): Promise<void> {
|
||||
const tx = txOpaque as Knex.Transaction;
|
||||
const { entityRef } = options;
|
||||
|
||||
const updateResult = await tx<DbRefreshStateRow>('refresh_state')
|
||||
.where({ entity_ref: entityRef.toLocaleLowerCase('en-US') })
|
||||
.update({ next_update_at: tx.fn.now() });
|
||||
if (updateResult === 0) {
|
||||
throw new NotFoundError(`Failed to schedule ${entityRef} for refresh`);
|
||||
}
|
||||
}
|
||||
|
||||
async refreshByRefreshKeys(
|
||||
txOpaque: Transaction,
|
||||
options: RefreshByKeyOptions,
|
||||
) {
|
||||
const tx = txOpaque as Knex.Transaction;
|
||||
const { keys } = options;
|
||||
|
||||
await tx<DbRefreshStateRow>('refresh_state')
|
||||
.whereIn('entity_id', function selectEntityRefs(tx2) {
|
||||
tx2
|
||||
.whereIn('key', keys)
|
||||
.select({
|
||||
entity_id: 'refresh_keys.entity_id',
|
||||
})
|
||||
.from('refresh_keys');
|
||||
})
|
||||
.update({ next_update_at: tx.fn.now() });
|
||||
}
|
||||
|
||||
async transaction<T>(fn: (tx: Transaction) => Promise<T>): Promise<T> {
|
||||
try {
|
||||
let result: T | undefined = undefined;
|
||||
@@ -597,123 +292,6 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Attempts to update an existing refresh state row, returning true if it was
|
||||
* updated and false if there was no entity with a matching ref and location key.
|
||||
*
|
||||
* Updating the entity will also cause it to be scheduled for immediate processing.
|
||||
*/
|
||||
private async updateUnprocessedEntity(
|
||||
tx: Knex.Transaction,
|
||||
entity: Entity,
|
||||
hash: string,
|
||||
locationKey?: string,
|
||||
): Promise<boolean> {
|
||||
const entityRef = stringifyEntityRef(entity);
|
||||
const serializedEntity = JSON.stringify(entity);
|
||||
|
||||
const refreshResult = await tx<DbRefreshStateRow>('refresh_state')
|
||||
.update({
|
||||
unprocessed_entity: serializedEntity,
|
||||
unprocessed_hash: hash,
|
||||
location_key: locationKey,
|
||||
last_discovery_at: tx.fn.now(),
|
||||
// We only get to this point if a processed entity actually had any changes, or
|
||||
// if an entity provider requested this mutation, meaning that we can safely
|
||||
// bump the deferred entities to the front of the queue for immediate processing.
|
||||
next_update_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');
|
||||
});
|
||||
|
||||
return refreshResult === 1;
|
||||
}
|
||||
|
||||
/**
|
||||
* Attempts to insert a new refresh state row for the given entity, returning
|
||||
* true if successful and false if there was a conflict.
|
||||
*/
|
||||
private async insertUnprocessedEntity(
|
||||
tx: Knex.Transaction,
|
||||
entity: Entity,
|
||||
hash: string,
|
||||
locationKey?: string,
|
||||
): Promise<boolean> {
|
||||
const entityRef = stringifyEntityRef(entity);
|
||||
const serializedEntity = JSON.stringify(entity);
|
||||
|
||||
try {
|
||||
let query = tx<DbRefreshStateRow>('refresh_state').insert({
|
||||
entity_id: uuid(),
|
||||
entity_ref: entityRef,
|
||||
unprocessed_entity: serializedEntity,
|
||||
unprocessed_hash: hash,
|
||||
errors: '',
|
||||
location_key: locationKey,
|
||||
next_update_at: tx.fn.now(),
|
||||
last_discovery_at: tx.fn.now(),
|
||||
});
|
||||
|
||||
// TODO(Rugvip): only tested towards MySQL, 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.includes('pg')) {
|
||||
query = query.onConflict('entity_ref').ignore() as any; // type here does not match runtime
|
||||
}
|
||||
|
||||
// Postgres gives as an object with rowCount, SQLite gives us an array
|
||||
const result: { rowCount?: number; length?: number } = await query;
|
||||
return result.rowCount === 1 || result.length === 1;
|
||||
} catch (error) {
|
||||
// SQLite, or MySQL reached this rather than the rowCount check above
|
||||
if (!isDatabaseConflictError(error)) {
|
||||
throw error;
|
||||
} else {
|
||||
this.options.logger.debug(
|
||||
`Unable to insert a new refresh state row, ${error}`,
|
||||
);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks whether a refresh state exists for the given entity that has a
|
||||
* location key that does not match the provided location key.
|
||||
*
|
||||
* @returns The conflicting key if there is one.
|
||||
*/
|
||||
private async checkLocationKeyConflict(
|
||||
tx: Knex.Transaction,
|
||||
entityRef: string,
|
||||
locationKey?: string,
|
||||
): Promise<string | undefined> {
|
||||
const row = await tx<DbRefreshStateRow>('refresh_state')
|
||||
.select('location_key')
|
||||
.where('entity_ref', entityRef)
|
||||
.first();
|
||||
|
||||
const conflictingKey = row?.location_key;
|
||||
|
||||
// If there's no existing key we can't have a conflict
|
||||
if (!conflictingKey) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
if (conflictingKey !== locationKey) {
|
||||
return conflictingKey;
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
private deduplicateRelations(rows: DbRelationsRow[]): DbRelationsRow[] {
|
||||
return lodash.uniqBy(
|
||||
rows,
|
||||
@@ -721,84 +299,6 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
|
||||
);
|
||||
}
|
||||
|
||||
private async createDelta(
|
||||
tx: Knex.Transaction,
|
||||
options: ReplaceUnprocessedEntitiesOptions,
|
||||
): Promise<{
|
||||
toAdd: { deferred: DeferredEntity; hash: string }[];
|
||||
toUpsert: { deferred: DeferredEntity; hash: string }[];
|
||||
toRemove: string[];
|
||||
}> {
|
||||
if (options.type === 'delta') {
|
||||
return {
|
||||
toAdd: [],
|
||||
toUpsert: options.added.map(e => ({
|
||||
deferred: e,
|
||||
hash: generateStableHash(e.entity),
|
||||
})),
|
||||
toRemove: options.removed.map(e => e.entityRef),
|
||||
};
|
||||
}
|
||||
|
||||
// Grab all of the existing references from the same source, and their locationKeys as well
|
||||
const oldRefs = await tx<DbRefreshStateReferencesRow>(
|
||||
'refresh_state_references',
|
||||
)
|
||||
.leftJoin<DbRefreshStateRow>('refresh_state', {
|
||||
target_entity_ref: 'entity_ref',
|
||||
})
|
||||
.where({ source_key: options.sourceKey })
|
||||
.select({
|
||||
target_entity_ref: 'refresh_state_references.target_entity_ref',
|
||||
location_key: 'refresh_state.location_key',
|
||||
unprocessed_hash: 'refresh_state.unprocessed_hash',
|
||||
});
|
||||
|
||||
const items = options.items.map(deferred => ({
|
||||
deferred,
|
||||
ref: stringifyEntityRef(deferred.entity),
|
||||
hash: generateStableHash(deferred.entity),
|
||||
}));
|
||||
|
||||
const oldRefsSet = new Map(
|
||||
oldRefs.map(r => [
|
||||
r.target_entity_ref,
|
||||
{
|
||||
locationKey: r.location_key,
|
||||
oldEntityHash: r.unprocessed_hash,
|
||||
},
|
||||
]),
|
||||
);
|
||||
const newRefsSet = new Set(items.map(item => item.ref));
|
||||
|
||||
const toAdd = new Array<{ deferred: DeferredEntity; hash: string }>();
|
||||
const toUpsert = new Array<{ deferred: DeferredEntity; hash: string }>();
|
||||
const toRemove = oldRefs
|
||||
.map(row => row.target_entity_ref)
|
||||
.filter(ref => !newRefsSet.has(ref));
|
||||
|
||||
for (const item of items) {
|
||||
const oldRef = oldRefsSet.get(item.ref);
|
||||
const upsertItem = { deferred: item.deferred, hash: item.hash };
|
||||
if (!oldRef) {
|
||||
// Add any entity that does not exist in the database
|
||||
toAdd.push(upsertItem);
|
||||
} 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);
|
||||
} else if (oldRef.oldEntityHash !== item.hash) {
|
||||
// Entities with modifications should be pushed through too
|
||||
toUpsert.push(upsertItem);
|
||||
}
|
||||
}
|
||||
|
||||
return { toAdd, toUpsert, toRemove };
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a set of deferred entities for processing.
|
||||
* The entities will be added at the front of the processing queue.
|
||||
@@ -822,23 +322,24 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
|
||||
const entityRef = stringifyEntityRef(entity);
|
||||
const hash = generateStableHash(entity);
|
||||
|
||||
const updated = await this.updateUnprocessedEntity(
|
||||
const updated = await updateUnprocessedEntity({
|
||||
tx,
|
||||
entity,
|
||||
hash,
|
||||
locationKey,
|
||||
);
|
||||
});
|
||||
if (updated) {
|
||||
stateReferences.push(entityRef);
|
||||
continue;
|
||||
}
|
||||
|
||||
const inserted = await this.insertUnprocessedEntity(
|
||||
const inserted = await insertUnprocessedEntity({
|
||||
tx,
|
||||
entity,
|
||||
hash,
|
||||
locationKey,
|
||||
);
|
||||
logger: this.options.logger,
|
||||
});
|
||||
if (inserted) {
|
||||
stateReferences.push(entityRef);
|
||||
continue;
|
||||
@@ -847,11 +348,11 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
|
||||
// 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 conflictingKey = await this.checkLocationKeyConflict(
|
||||
const conflictingKey = await checkLocationKeyConflict({
|
||||
tx,
|
||||
entityRef,
|
||||
locationKey,
|
||||
);
|
||||
});
|
||||
if (conflictingKey) {
|
||||
this.options.logger.warn(
|
||||
`Detected conflicting entityRef ${entityRef} already referenced by ${conflictingKey} and now also ${locationKey}`,
|
||||
|
||||
@@ -0,0 +1,715 @@
|
||||
/*
|
||||
* 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 { getVoidLogger } from '@backstage/backend-common';
|
||||
import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils';
|
||||
import { Entity } from '@backstage/catalog-model';
|
||||
import { Knex } from 'knex';
|
||||
import * as uuid from 'uuid';
|
||||
import { Logger } from 'winston';
|
||||
import { DefaultProviderDatabase } from './DefaultProviderDatabase';
|
||||
import { applyDatabaseMigrations } from './migrations';
|
||||
import { DbRefreshStateReferencesRow, DbRefreshStateRow } from './tables';
|
||||
|
||||
describe('DefaultProviderDatabase', () => {
|
||||
const defaultLogger = getVoidLogger();
|
||||
const databases = TestDatabases.create({
|
||||
ids: ['MYSQL_8', 'POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'],
|
||||
});
|
||||
|
||||
async function createDatabase(
|
||||
databaseId: TestDatabaseId,
|
||||
logger: Logger = defaultLogger,
|
||||
) {
|
||||
const knex = await databases.init(databaseId);
|
||||
await applyDatabaseMigrations(knex);
|
||||
return {
|
||||
knex,
|
||||
db: new DefaultProviderDatabase({
|
||||
database: knex,
|
||||
logger,
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
const insertRefRow = async (db: Knex, ref: DbRefreshStateReferencesRow) => {
|
||||
return db<DbRefreshStateReferencesRow>('refresh_state_references').insert(
|
||||
ref,
|
||||
);
|
||||
};
|
||||
|
||||
const insertRefreshStateRow = async (db: Knex, ref: DbRefreshStateRow) => {
|
||||
await db<DbRefreshStateRow>('refresh_state').insert(ref);
|
||||
};
|
||||
|
||||
describe('replaceUnprocessedEntities', () => {
|
||||
const createLocations = async (db: Knex, entityRefs: string[]) => {
|
||||
for (const ref of entityRefs) {
|
||||
await insertRefreshStateRow(db, {
|
||||
entity_id: uuid.v4(),
|
||||
entity_ref: ref,
|
||||
unprocessed_entity: '{}',
|
||||
processed_entity: '{}',
|
||||
errors: '[]',
|
||||
next_update_at: '2021-04-01 13:37:00',
|
||||
last_discovery_at: '2021-04-01 13:37:00',
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'replaces all existing state correctly for simple dependency chains, %p',
|
||||
async databaseId => {
|
||||
const { knex, db } = await createDatabase(databaseId);
|
||||
/*
|
||||
config -> location:default/root -> location:default/root-1 -> location:default/root-2
|
||||
database -> location:default/second -> location:default/root-2
|
||||
*/
|
||||
await createLocations(knex, [
|
||||
'location:default/root',
|
||||
'location:default/root-1',
|
||||
'location:default/root-2',
|
||||
'location:default/second',
|
||||
]);
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_key: 'config',
|
||||
target_entity_ref: 'location:default/root',
|
||||
});
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_key: 'database',
|
||||
target_entity_ref: 'location:default/second',
|
||||
});
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_entity_ref: 'location:default/root',
|
||||
target_entity_ref: 'location:default/root-1',
|
||||
});
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_entity_ref: 'location:default/root-1',
|
||||
target_entity_ref: 'location:default/root-2',
|
||||
});
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_entity_ref: 'location:default/second',
|
||||
target_entity_ref: 'location:default/root-2',
|
||||
});
|
||||
|
||||
await db.transaction(tx =>
|
||||
db.replaceUnprocessedEntities(tx, {
|
||||
type: 'full',
|
||||
sourceKey: 'config',
|
||||
items: [
|
||||
{
|
||||
entity: {
|
||||
apiVersion: '1.0.0',
|
||||
metadata: {
|
||||
name: 'new-root',
|
||||
},
|
||||
kind: 'Location',
|
||||
} as Entity,
|
||||
locationKey: 'file:///tmp/foobar',
|
||||
},
|
||||
],
|
||||
}),
|
||||
);
|
||||
|
||||
const currentRefreshState = await knex<DbRefreshStateRow>(
|
||||
'refresh_state',
|
||||
).select();
|
||||
|
||||
const currentRefRowState = await knex<DbRefreshStateReferencesRow>(
|
||||
'refresh_state_references',
|
||||
).select();
|
||||
|
||||
for (const ref of [
|
||||
'location:default/root',
|
||||
'location:default/root-1',
|
||||
]) {
|
||||
expect(
|
||||
currentRefreshState.some(t => t.entity_ref === ref),
|
||||
).toBeFalsy();
|
||||
}
|
||||
|
||||
expect(
|
||||
currentRefreshState.some(
|
||||
t => t.entity_ref === 'location:default/new-root',
|
||||
),
|
||||
).toBeTruthy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.source_entity_ref === 'location:default/root' &&
|
||||
t.target_entity_ref === 'location:default/root-1',
|
||||
),
|
||||
).toBeFalsy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.source_entity_ref === 'location:default/root-1' &&
|
||||
t.target_entity_ref === 'location:default/root-2',
|
||||
),
|
||||
).toBeFalsy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.target_entity_ref === 'location:default/root-1' &&
|
||||
t.source_key === 'config',
|
||||
),
|
||||
).toBeFalsy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.target_entity_ref === 'location:default/new-root' &&
|
||||
t.source_key === 'config',
|
||||
),
|
||||
).toBeTruthy();
|
||||
},
|
||||
60_000,
|
||||
);
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should work for more complex chains, %p',
|
||||
async databaseId => {
|
||||
const { knex, db } = await createDatabase(databaseId);
|
||||
/*
|
||||
config -> location:default/root -> location:default/root-1 -> location:default/root-2
|
||||
config -> location:default/root -> location:default/root-1a -> location:default/root-2
|
||||
*/
|
||||
await createLocations(knex, [
|
||||
'location:default/root',
|
||||
'location:default/root-1',
|
||||
'location:default/root-2',
|
||||
'location:default/root-1a',
|
||||
]);
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_key: 'config',
|
||||
target_entity_ref: 'location:default/root',
|
||||
});
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_entity_ref: 'location:default/root',
|
||||
target_entity_ref: 'location:default/root-1',
|
||||
});
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_entity_ref: 'location:default/root',
|
||||
target_entity_ref: 'location:default/root-1a',
|
||||
});
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_entity_ref: 'location:default/root-1',
|
||||
target_entity_ref: 'location:default/root-2',
|
||||
});
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_entity_ref: 'location:default/root-1a',
|
||||
target_entity_ref: 'location:default/root-2',
|
||||
});
|
||||
|
||||
await db.transaction(async tx => {
|
||||
await db.replaceUnprocessedEntities(tx, {
|
||||
type: 'full',
|
||||
sourceKey: 'config',
|
||||
items: [
|
||||
{
|
||||
entity: {
|
||||
apiVersion: '1.0.0',
|
||||
metadata: {
|
||||
name: 'new-root',
|
||||
},
|
||||
kind: 'Location',
|
||||
} as Entity,
|
||||
locationKey: 'file:/tmp/foobar',
|
||||
},
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
const currentRefreshState = await knex<DbRefreshStateRow>(
|
||||
'refresh_state',
|
||||
).select();
|
||||
|
||||
const currentRefRowState = await knex<DbRefreshStateReferencesRow>(
|
||||
'refresh_state_references',
|
||||
).select();
|
||||
|
||||
const deletedRefs = [
|
||||
'location:default/root',
|
||||
'location:default/root-1',
|
||||
'location:default/root-1a',
|
||||
'location:default/root-2',
|
||||
];
|
||||
|
||||
for (const ref of deletedRefs) {
|
||||
expect(
|
||||
currentRefreshState.some(t => t.entity_ref === ref),
|
||||
).toBeFalsy();
|
||||
}
|
||||
|
||||
expect(
|
||||
currentRefreshState.some(
|
||||
t => t.entity_ref === 'location:default/new-root',
|
||||
),
|
||||
).toBeTruthy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.source_key === 'config' &&
|
||||
t.target_entity_ref === 'location:default/new-root',
|
||||
),
|
||||
).toBeTruthy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.source_key === 'config' &&
|
||||
t.target_entity_ref === 'location:default/root',
|
||||
),
|
||||
).toBeFalsy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.source_entity_ref === 'location:default/root' &&
|
||||
t.target_entity_ref === 'location:default/root-1',
|
||||
),
|
||||
).toBeFalsy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.source_entity_ref === 'location:default/root' &&
|
||||
t.target_entity_ref === 'location:default/root-1a',
|
||||
),
|
||||
).toBeFalsy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.source_entity_ref === 'location:default/root-1' &&
|
||||
t.target_entity_ref === 'location:default/root-2',
|
||||
),
|
||||
).toBeFalsy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.source_entity_ref === 'location:default/root-1a' &&
|
||||
t.target_entity_ref === 'location:default/root-2',
|
||||
),
|
||||
).toBeFalsy();
|
||||
},
|
||||
60_000,
|
||||
);
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should add new locations using the delta options, %p',
|
||||
async databaseId => {
|
||||
const { knex, db } = await createDatabase(databaseId);
|
||||
|
||||
// Existing state and references should stay
|
||||
await createLocations(knex, ['location:default/existing']);
|
||||
await insertRefRow(knex, {
|
||||
source_key: 'lols',
|
||||
target_entity_ref: 'location:default/existing',
|
||||
});
|
||||
|
||||
await db.transaction(async tx => {
|
||||
await db.replaceUnprocessedEntities(tx, {
|
||||
type: 'delta',
|
||||
sourceKey: 'lols',
|
||||
removed: [],
|
||||
added: [
|
||||
{
|
||||
entity: {
|
||||
apiVersion: '1.0.0',
|
||||
metadata: {
|
||||
name: 'new-root',
|
||||
},
|
||||
kind: 'Location',
|
||||
} as Entity,
|
||||
locationKey: 'file:///tmp/foobar',
|
||||
},
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
const currentRefreshState = await knex<DbRefreshStateRow>(
|
||||
'refresh_state',
|
||||
).select();
|
||||
|
||||
const currentRefRowState = await knex<DbRefreshStateReferencesRow>(
|
||||
'refresh_state_references',
|
||||
).select();
|
||||
|
||||
expect(
|
||||
currentRefreshState.some(
|
||||
t => t.entity_ref === 'location:default/new-root',
|
||||
),
|
||||
).toBeTruthy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.source_key === 'lols' &&
|
||||
t.target_entity_ref === 'location:default/new-root',
|
||||
),
|
||||
).toBeTruthy();
|
||||
|
||||
expect(
|
||||
currentRefreshState.some(
|
||||
t => t.entity_ref === 'location:default/existing',
|
||||
),
|
||||
).toBeTruthy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.source_key === 'lols' &&
|
||||
t.target_entity_ref === 'location:default/existing',
|
||||
),
|
||||
).toBeTruthy();
|
||||
},
|
||||
60_000,
|
||||
);
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should not remove locations that are referenced elsewhere, %p',
|
||||
async databaseId => {
|
||||
const { knex, db } = await createDatabase(databaseId);
|
||||
/*
|
||||
config-1 -> location:default/root
|
||||
config-2 -> location:default/root
|
||||
*/
|
||||
await createLocations(knex, ['location:default/root']);
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_key: 'config-1',
|
||||
target_entity_ref: 'location:default/root',
|
||||
});
|
||||
await insertRefRow(knex, {
|
||||
source_key: 'config-2',
|
||||
target_entity_ref: 'location:default/root',
|
||||
});
|
||||
|
||||
await db.transaction(async tx => {
|
||||
await db.replaceUnprocessedEntities(tx, {
|
||||
type: 'full',
|
||||
sourceKey: 'config-1',
|
||||
items: [],
|
||||
});
|
||||
});
|
||||
|
||||
const currentRefreshState = await knex<DbRefreshStateRow>(
|
||||
'refresh_state',
|
||||
).select();
|
||||
|
||||
const currentRefRowState = await knex<DbRefreshStateReferencesRow>(
|
||||
'refresh_state_references',
|
||||
).select();
|
||||
|
||||
expect(currentRefRowState).toEqual([
|
||||
expect.objectContaining({
|
||||
source_key: 'config-2',
|
||||
target_entity_ref: 'location:default/root',
|
||||
}),
|
||||
]);
|
||||
|
||||
expect(currentRefreshState).toEqual([
|
||||
expect.objectContaining({
|
||||
entity_ref: 'location:default/root',
|
||||
}),
|
||||
]);
|
||||
},
|
||||
60_000,
|
||||
);
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should remove old locations using the delta options, %p',
|
||||
async databaseId => {
|
||||
const { knex, db } = await createDatabase(databaseId);
|
||||
await createLocations(knex, ['location:default/new-root']);
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_key: 'lols',
|
||||
target_entity_ref: 'location:default/new-root',
|
||||
});
|
||||
|
||||
await db.transaction(async tx => {
|
||||
await db.replaceUnprocessedEntities(tx, {
|
||||
type: 'delta',
|
||||
sourceKey: 'lols',
|
||||
added: [],
|
||||
removed: [
|
||||
{
|
||||
entityRef: 'location:default/new-root',
|
||||
locationKey: 'file:/tmp/foobar',
|
||||
},
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
const currentRefreshState = await knex<DbRefreshStateRow>(
|
||||
'refresh_state',
|
||||
).select();
|
||||
|
||||
const currentRefRowState = await knex<DbRefreshStateReferencesRow>(
|
||||
'refresh_state_references',
|
||||
).select();
|
||||
|
||||
expect(
|
||||
currentRefreshState.some(
|
||||
t => t.entity_ref === 'location:default/new-root',
|
||||
),
|
||||
).toBeFalsy();
|
||||
|
||||
expect(
|
||||
currentRefRowState.some(
|
||||
t =>
|
||||
t.source_key === 'lols' &&
|
||||
t.target_entity_ref === 'location:default/new-root',
|
||||
),
|
||||
).toBeFalsy();
|
||||
},
|
||||
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,
|
||||
);
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should support replacing modified entities during a full update, %p',
|
||||
async databaseId => {
|
||||
const { knex, db } = await createDatabase(databaseId);
|
||||
|
||||
await db.transaction(async tx => {
|
||||
await db.replaceUnprocessedEntities(tx, {
|
||||
type: 'full',
|
||||
sourceKey: 'lols',
|
||||
items: [
|
||||
{
|
||||
entity: {
|
||||
apiVersion: '1',
|
||||
kind: 'Component',
|
||||
metadata: { name: 'a' },
|
||||
spec: { marker: 'WILL_CHANGE' },
|
||||
} as Entity,
|
||||
locationKey: 'file:///tmp/a',
|
||||
},
|
||||
{
|
||||
entity: {
|
||||
apiVersion: '1',
|
||||
kind: 'Component',
|
||||
metadata: { name: 'b' },
|
||||
spec: { marker: 'NEVER_CHANGES' },
|
||||
} as Entity,
|
||||
locationKey: 'file:///tmp/b',
|
||||
},
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
let state = await knex<DbRefreshStateRow>('refresh_state').select();
|
||||
expect(state).toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
entity_ref: 'component:default/a',
|
||||
location_key: 'file:///tmp/a',
|
||||
unprocessed_entity: expect.stringContaining('WILL_CHANGE'),
|
||||
}),
|
||||
expect.objectContaining({
|
||||
entity_ref: 'component:default/b',
|
||||
location_key: 'file:///tmp/b',
|
||||
unprocessed_entity: expect.stringContaining('NEVER_CHANGES'),
|
||||
}),
|
||||
]),
|
||||
);
|
||||
|
||||
await db.transaction(async tx => {
|
||||
await db.replaceUnprocessedEntities(tx, {
|
||||
type: 'full',
|
||||
sourceKey: 'lols',
|
||||
items: [
|
||||
{
|
||||
entity: {
|
||||
apiVersion: '1',
|
||||
kind: 'Component',
|
||||
metadata: { name: 'a' },
|
||||
spec: { marker: 'HAS_CHANGED' },
|
||||
} as Entity,
|
||||
locationKey: 'file:///tmp/a',
|
||||
},
|
||||
{
|
||||
entity: {
|
||||
apiVersion: '1',
|
||||
kind: 'Component',
|
||||
metadata: { name: 'b' },
|
||||
spec: { marker: 'NEVER_CHANGES' },
|
||||
} as Entity,
|
||||
locationKey: 'file:///tmp/b',
|
||||
},
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
state = await knex<DbRefreshStateRow>('refresh_state').select();
|
||||
expect(state).toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
entity_ref: 'component:default/a',
|
||||
location_key: 'file:///tmp/a',
|
||||
unprocessed_entity: expect.stringContaining('HAS_CHANGED'),
|
||||
}),
|
||||
expect.objectContaining({
|
||||
entity_ref: 'component:default/b',
|
||||
location_key: 'file:///tmp/b',
|
||||
unprocessed_entity: expect.stringContaining('NEVER_CHANGES'),
|
||||
}),
|
||||
]),
|
||||
);
|
||||
},
|
||||
60_000,
|
||||
);
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should successfully fall back from batch to individual mode on conflicts, %p',
|
||||
async databaseId => {
|
||||
const fakeLogger = {
|
||||
debug: jest.fn(),
|
||||
};
|
||||
const { knex, db } = await createDatabase(
|
||||
databaseId,
|
||||
fakeLogger as any,
|
||||
);
|
||||
|
||||
await createLocations(knex, ['component:default/a']);
|
||||
|
||||
await insertRefRow(knex, {
|
||||
source_key: undefined,
|
||||
target_entity_ref: 'component:default/a',
|
||||
});
|
||||
|
||||
await db.transaction(async tx => {
|
||||
await db.replaceUnprocessedEntities(tx, {
|
||||
type: 'full',
|
||||
sourceKey: 'lols',
|
||||
items: [
|
||||
{
|
||||
entity: {
|
||||
apiVersion: '1',
|
||||
kind: 'Component',
|
||||
metadata: { name: 'a' },
|
||||
spec: { marker: 'WILL_CHANGE' },
|
||||
} as Entity,
|
||||
locationKey: 'file:///tmp/a',
|
||||
},
|
||||
],
|
||||
});
|
||||
});
|
||||
expect(fakeLogger.debug).toHaveBeenCalledWith(
|
||||
expect.stringMatching(
|
||||
/Fast insert path failed, falling back to slow path/,
|
||||
),
|
||||
);
|
||||
|
||||
const state = await knex<DbRefreshStateRow>('refresh_state').select();
|
||||
expect(state).toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
entity_ref: 'component:default/a',
|
||||
location_key: 'file:///tmp/a',
|
||||
unprocessed_entity: expect.stringContaining('WILL_CHANGE'),
|
||||
}),
|
||||
]),
|
||||
);
|
||||
},
|
||||
60_000,
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,277 @@
|
||||
/*
|
||||
* 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 { isDatabaseConflictError } from '@backstage/backend-common';
|
||||
import { stringifyEntityRef } from '@backstage/catalog-model';
|
||||
import { DeferredEntity } from '@backstage/plugin-catalog-node';
|
||||
import { Knex } from 'knex';
|
||||
import lodash from 'lodash';
|
||||
import { v4 as uuid } from 'uuid';
|
||||
import type { Logger } from 'winston';
|
||||
import { rethrowError } from './conversion';
|
||||
import { deleteWithEagerPruningOfChildren } from './operations/provider/deleteWithEagerPruningOfChildren';
|
||||
import { refreshByRefreshKeys } from './operations/provider/refreshByRefreshKeys';
|
||||
import { checkLocationKeyConflict } from './operations/refreshState/checkLocationKeyConflict';
|
||||
import { insertUnprocessedEntity } from './operations/refreshState/insertUnprocessedEntity';
|
||||
import { updateUnprocessedEntity } from './operations/refreshState/updateUnprocessedEntity';
|
||||
import { DbRefreshStateReferencesRow, DbRefreshStateRow } from './tables';
|
||||
import {
|
||||
ProviderDatabase,
|
||||
RefreshByKeyOptions,
|
||||
ReplaceUnprocessedEntitiesOptions,
|
||||
Transaction,
|
||||
} from './types';
|
||||
import { generateStableHash } from './util';
|
||||
|
||||
// The number of items that are sent per batch to the database layer, when
|
||||
// doing .batchInsert calls to knex. This needs to be low enough to not cause
|
||||
// errors in the underlying engine due to exceeding query limits, but large
|
||||
// enough to get the speed benefits.
|
||||
const BATCH_SIZE = 50;
|
||||
|
||||
export class DefaultProviderDatabase implements ProviderDatabase {
|
||||
constructor(
|
||||
private readonly options: {
|
||||
database: Knex;
|
||||
logger: Logger;
|
||||
},
|
||||
) {}
|
||||
|
||||
async transaction<T>(fn: (tx: Transaction) => Promise<T>): Promise<T> {
|
||||
try {
|
||||
let result: T | undefined = undefined;
|
||||
await this.options.database.transaction(
|
||||
async tx => {
|
||||
// We can't return here, as knex swallows the return type in case the
|
||||
// transaction is rolled back:
|
||||
// https://github.com/knex/knex/blob/e37aeaa31c8ef9c1b07d2e4d3ec6607e557d800d/lib/transaction.js#L136
|
||||
result = await fn(tx);
|
||||
},
|
||||
{
|
||||
// If we explicitly trigger a rollback, don't fail.
|
||||
doNotRejectOnRollback: true,
|
||||
},
|
||||
);
|
||||
return result!;
|
||||
} catch (e) {
|
||||
this.options.logger.debug(`Error during transaction, ${e}`);
|
||||
throw rethrowError(e);
|
||||
}
|
||||
}
|
||||
|
||||
async replaceUnprocessedEntities(
|
||||
txOpaque: Transaction,
|
||||
options: ReplaceUnprocessedEntitiesOptions,
|
||||
): Promise<void> {
|
||||
const tx = txOpaque as Knex.Transaction;
|
||||
const { toAdd, toUpsert, toRemove } = await this.createDelta(tx, options);
|
||||
|
||||
if (toRemove.length) {
|
||||
const removedCount = await deleteWithEagerPruningOfChildren({
|
||||
tx,
|
||||
entityRefs: toRemove,
|
||||
sourceKey: options.sourceKey,
|
||||
});
|
||||
this.options.logger.debug(
|
||||
`removed, ${removedCount} entities: ${JSON.stringify(toRemove)}`,
|
||||
);
|
||||
}
|
||||
|
||||
if (toAdd.length) {
|
||||
// The reason for this chunking, rather than just massively batch
|
||||
// inserting the entire payload, is that we fall back to the individual
|
||||
// upsert mechanism below on conflicts. That path is massively slower than
|
||||
// the fast batch path, so we don't want to end up accidentally having to
|
||||
// for example item-by-item upsert tens of thousands of entities in a
|
||||
// large initial delivery dump. The implication is that the size of these
|
||||
// chunks needs to weigh the benefit of fast successful inserts, against
|
||||
// the drawback of super slow but more rare fallbacks. There's quickly
|
||||
// diminishing returns though with turning up this value way high.
|
||||
for (const chunk of lodash.chunk(toAdd, 50)) {
|
||||
try {
|
||||
await tx.batchInsert(
|
||||
'refresh_state',
|
||||
chunk.map(item => ({
|
||||
entity_id: uuid(),
|
||||
entity_ref: stringifyEntityRef(item.deferred.entity),
|
||||
unprocessed_entity: JSON.stringify(item.deferred.entity),
|
||||
unprocessed_hash: item.hash,
|
||||
errors: '',
|
||||
location_key: item.deferred.locationKey,
|
||||
next_update_at: tx.fn.now(),
|
||||
last_discovery_at: tx.fn.now(),
|
||||
})),
|
||||
BATCH_SIZE,
|
||||
);
|
||||
await tx.batchInsert(
|
||||
'refresh_state_references',
|
||||
chunk.map(item => ({
|
||||
source_key: options.sourceKey,
|
||||
target_entity_ref: stringifyEntityRef(item.deferred.entity),
|
||||
})),
|
||||
BATCH_SIZE,
|
||||
);
|
||||
} catch (error) {
|
||||
if (!isDatabaseConflictError(error)) {
|
||||
throw error;
|
||||
} else {
|
||||
this.options.logger.debug(
|
||||
`Fast insert path failed, falling back to slow path, ${error}`,
|
||||
);
|
||||
toUpsert.push(...chunk);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (toUpsert.length) {
|
||||
for (const {
|
||||
deferred: { entity, locationKey },
|
||||
hash,
|
||||
} of toUpsert) {
|
||||
const entityRef = stringifyEntityRef(entity);
|
||||
|
||||
try {
|
||||
let ok = await updateUnprocessedEntity({
|
||||
tx,
|
||||
entity,
|
||||
hash,
|
||||
locationKey,
|
||||
});
|
||||
if (!ok) {
|
||||
ok = await insertUnprocessedEntity({
|
||||
tx,
|
||||
entity,
|
||||
hash,
|
||||
locationKey,
|
||||
logger: this.options.logger,
|
||||
});
|
||||
}
|
||||
|
||||
if (ok) {
|
||||
await tx<DbRefreshStateReferencesRow>(
|
||||
'refresh_state_references',
|
||||
).insert({
|
||||
source_key: options.sourceKey,
|
||||
target_entity_ref: entityRef,
|
||||
});
|
||||
} else {
|
||||
const conflictingKey = await checkLocationKeyConflict({
|
||||
tx,
|
||||
entityRef,
|
||||
locationKey,
|
||||
});
|
||||
if (conflictingKey) {
|
||||
this.options.logger.warn(
|
||||
`Source ${options.sourceKey} detected conflicting entityRef ${entityRef} already referenced by ${conflictingKey} and now also ${locationKey}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
this.options.logger.error(
|
||||
`Failed to add '${entityRef}' from source '${options.sourceKey}', ${error}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async refreshByRefreshKeys(
|
||||
txOpaque: Transaction,
|
||||
options: RefreshByKeyOptions,
|
||||
) {
|
||||
const tx = txOpaque as Knex.Transaction;
|
||||
await refreshByRefreshKeys({ tx, keys: options.keys });
|
||||
}
|
||||
|
||||
private async createDelta(
|
||||
tx: Knex.Transaction,
|
||||
options: ReplaceUnprocessedEntitiesOptions,
|
||||
): Promise<{
|
||||
toAdd: { deferred: DeferredEntity; hash: string }[];
|
||||
toUpsert: { deferred: DeferredEntity; hash: string }[];
|
||||
toRemove: string[];
|
||||
}> {
|
||||
if (options.type === 'delta') {
|
||||
return {
|
||||
toAdd: [],
|
||||
toUpsert: options.added.map(e => ({
|
||||
deferred: e,
|
||||
hash: generateStableHash(e.entity),
|
||||
})),
|
||||
toRemove: options.removed.map(e => e.entityRef),
|
||||
};
|
||||
}
|
||||
|
||||
// Grab all of the existing references from the same source, and their locationKeys as well
|
||||
const oldRefs = await tx<DbRefreshStateReferencesRow>(
|
||||
'refresh_state_references',
|
||||
)
|
||||
.leftJoin<DbRefreshStateRow>('refresh_state', {
|
||||
target_entity_ref: 'entity_ref',
|
||||
})
|
||||
.where({ source_key: options.sourceKey })
|
||||
.select({
|
||||
target_entity_ref: 'refresh_state_references.target_entity_ref',
|
||||
location_key: 'refresh_state.location_key',
|
||||
unprocessed_hash: 'refresh_state.unprocessed_hash',
|
||||
});
|
||||
|
||||
const items = options.items.map(deferred => ({
|
||||
deferred,
|
||||
ref: stringifyEntityRef(deferred.entity),
|
||||
hash: generateStableHash(deferred.entity),
|
||||
}));
|
||||
|
||||
const oldRefsSet = new Map(
|
||||
oldRefs.map(r => [
|
||||
r.target_entity_ref,
|
||||
{
|
||||
locationKey: r.location_key,
|
||||
oldEntityHash: r.unprocessed_hash,
|
||||
},
|
||||
]),
|
||||
);
|
||||
const newRefsSet = new Set(items.map(item => item.ref));
|
||||
|
||||
const toAdd = new Array<{ deferred: DeferredEntity; hash: string }>();
|
||||
const toUpsert = new Array<{ deferred: DeferredEntity; hash: string }>();
|
||||
const toRemove = oldRefs
|
||||
.map(row => row.target_entity_ref)
|
||||
.filter(ref => !newRefsSet.has(ref));
|
||||
|
||||
for (const item of items) {
|
||||
const oldRef = oldRefsSet.get(item.ref);
|
||||
const upsertItem = { deferred: item.deferred, hash: item.hash };
|
||||
if (!oldRef) {
|
||||
// Add any entity that does not exist in the database
|
||||
toAdd.push(upsertItem);
|
||||
} 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);
|
||||
} else if (oldRef.oldEntityHash !== item.hash) {
|
||||
// Entities with modifications should be pushed through too
|
||||
toUpsert.push(upsertItem);
|
||||
}
|
||||
}
|
||||
|
||||
return { toAdd, toUpsert, toRemove };
|
||||
}
|
||||
}
|
||||
+155
@@ -0,0 +1,155 @@
|
||||
/*
|
||||
* 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 lodash from 'lodash';
|
||||
import { DbRefreshStateReferencesRow, DbRefreshStateRow } from '../../tables';
|
||||
|
||||
/**
|
||||
* Given a number of entity refs originally created by a given entity provider
|
||||
* (source key), remove those entities from the refresh state, and at the same
|
||||
* time recursively remove every child that is a direct or indirect result of
|
||||
* processing those entities, if they would have otherwise become orphaned by
|
||||
* the removal of their parents.
|
||||
*/
|
||||
export async function deleteWithEagerPruningOfChildren(options: {
|
||||
tx: Knex.Transaction;
|
||||
entityRefs: string[];
|
||||
sourceKey: string;
|
||||
}): Promise<number> {
|
||||
const { tx, entityRefs, sourceKey } = options;
|
||||
let removedCount = 0;
|
||||
|
||||
const rootId = () =>
|
||||
tx.raw(
|
||||
tx.client.config.client.includes('mysql')
|
||||
? 'CAST(NULL as UNSIGNED INT)'
|
||||
: 'CAST(NULL as INT)',
|
||||
[],
|
||||
);
|
||||
|
||||
// Split up the operation by (large) chunks, so that we do not hit database
|
||||
// limits for the number of permitted bindings on a precompiled statement
|
||||
for (const refs of lodash.chunk(entityRefs, 1000)) {
|
||||
/*
|
||||
WITH RECURSIVE
|
||||
-- All the nodes that can be reached downwards from our root
|
||||
descendants(root_id, entity_ref) AS (
|
||||
SELECT id, target_entity_ref
|
||||
FROM refresh_state_references
|
||||
WHERE source_key = "R1" AND target_entity_ref = "A"
|
||||
UNION
|
||||
SELECT descendants.root_id, target_entity_ref
|
||||
FROM descendants
|
||||
JOIN refresh_state_references ON source_entity_ref = descendants.entity_ref
|
||||
),
|
||||
-- All the nodes that can be reached upwards from the descendants
|
||||
ancestors(root_id, via_entity_ref, to_entity_ref) AS (
|
||||
SELECT CAST(NULL as INT), entity_ref, entity_ref
|
||||
FROM descendants
|
||||
UNION
|
||||
SELECT
|
||||
CASE WHEN source_key IS NOT NULL THEN id ELSE NULL END,
|
||||
source_entity_ref,
|
||||
ancestors.to_entity_ref
|
||||
FROM ancestors
|
||||
JOIN refresh_state_references ON target_entity_ref = ancestors.via_entity_ref
|
||||
)
|
||||
-- Start out with all of the descendants
|
||||
SELECT descendants.entity_ref
|
||||
FROM descendants
|
||||
-- Expand with all ancestors that point to those, but aren't the current root
|
||||
LEFT OUTER JOIN ancestors
|
||||
ON ancestors.to_entity_ref = descendants.entity_ref
|
||||
AND ancestors.root_id IS NOT NULL
|
||||
AND ancestors.root_id != descendants.root_id
|
||||
-- Exclude all lines that had such a foreign ancestor
|
||||
WHERE ancestors.root_id IS NULL;
|
||||
*/
|
||||
removedCount += await tx<DbRefreshStateRow>('refresh_state')
|
||||
.whereIn('entity_ref', function orphanedEntityRefs(orphans) {
|
||||
return (
|
||||
orphans
|
||||
// All the nodes that can be reached downwards from our root
|
||||
.withRecursive('descendants', function descendants(outer) {
|
||||
return outer
|
||||
.select({ root_id: 'id', entity_ref: 'target_entity_ref' })
|
||||
.from('refresh_state_references')
|
||||
.where('source_key', sourceKey)
|
||||
.whereIn('target_entity_ref', refs)
|
||||
.union(function recursive(inner) {
|
||||
return inner
|
||||
.select({
|
||||
root_id: 'descendants.root_id',
|
||||
entity_ref: 'refresh_state_references.target_entity_ref',
|
||||
})
|
||||
.from('descendants')
|
||||
.join('refresh_state_references', {
|
||||
'descendants.entity_ref':
|
||||
'refresh_state_references.source_entity_ref',
|
||||
});
|
||||
});
|
||||
})
|
||||
// All the nodes that can be reached upwards from the descendants
|
||||
.withRecursive('ancestors', function ancestors(outer) {
|
||||
return outer
|
||||
.select({
|
||||
root_id: rootId(),
|
||||
via_entity_ref: 'entity_ref',
|
||||
to_entity_ref: 'entity_ref',
|
||||
})
|
||||
.from('descendants')
|
||||
.union(function recursive(inner) {
|
||||
return inner
|
||||
.select({
|
||||
root_id: tx.raw(
|
||||
'CASE WHEN source_key IS NOT NULL THEN id ELSE NULL END',
|
||||
[],
|
||||
),
|
||||
via_entity_ref: 'source_entity_ref',
|
||||
to_entity_ref: 'ancestors.to_entity_ref',
|
||||
})
|
||||
.from('ancestors')
|
||||
.join('refresh_state_references', {
|
||||
target_entity_ref: 'ancestors.via_entity_ref',
|
||||
});
|
||||
});
|
||||
})
|
||||
// Start out with all of the descendants
|
||||
.select('descendants.entity_ref')
|
||||
.from('descendants')
|
||||
// Expand with all ancestors that point to those, but aren't the current root
|
||||
.leftOuterJoin('ancestors', function keepaliveRoots() {
|
||||
this.on('ancestors.to_entity_ref', '=', 'descendants.entity_ref');
|
||||
this.andOnNotNull('ancestors.root_id');
|
||||
this.andOn('ancestors.root_id', '!=', 'descendants.root_id');
|
||||
})
|
||||
.whereNull('ancestors.root_id')
|
||||
);
|
||||
})
|
||||
.delete();
|
||||
|
||||
// Delete the references that originate only from this entity provider. Note
|
||||
// that there may be more than one entity provider making a "claim" for a
|
||||
// given root entity, if they emit with the same location key.
|
||||
await tx<DbRefreshStateReferencesRow>('refresh_state_references')
|
||||
.where('source_key', '=', sourceKey)
|
||||
.whereIn('target_entity_ref', refs)
|
||||
.delete();
|
||||
}
|
||||
|
||||
return removedCount;
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
/*
|
||||
* 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 { DbRefreshStateRow } from '../../tables';
|
||||
|
||||
/**
|
||||
* Schedules a future refresh of entities, by so called "refresh keys" that may
|
||||
* be associated with one or more entities. Note that this does not mean that
|
||||
* the refresh happens immediately, but rather that their scheduling time gets
|
||||
* moved up the queue and will get picked up eventually by the regular
|
||||
* processing loop.
|
||||
*/
|
||||
export async function refreshByRefreshKeys(options: {
|
||||
tx: Knex.Transaction;
|
||||
keys: string[];
|
||||
}): Promise<void> {
|
||||
const { tx, keys } = options;
|
||||
|
||||
await tx<DbRefreshStateRow>('refresh_state')
|
||||
.whereIn('entity_id', function selectEntityRefs(inner) {
|
||||
inner
|
||||
.whereIn('key', keys)
|
||||
.select({
|
||||
entity_id: 'refresh_keys.entity_id',
|
||||
})
|
||||
.from('refresh_keys');
|
||||
})
|
||||
.update({ next_update_at: tx.fn.now() });
|
||||
}
|
||||
+49
@@ -0,0 +1,49 @@
|
||||
/*
|
||||
* 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 { DbRefreshStateRow } from '../../tables';
|
||||
|
||||
/**
|
||||
* Checks whether a refresh state exists for the given entity that has a
|
||||
* location key that does not match the provided location key.
|
||||
*
|
||||
* @returns The conflicting key if there is one.
|
||||
*/
|
||||
export async function checkLocationKeyConflict(options: {
|
||||
tx: Knex.Transaction;
|
||||
entityRef: string;
|
||||
locationKey?: string;
|
||||
}): Promise<string | undefined> {
|
||||
const { tx, entityRef, locationKey } = options;
|
||||
|
||||
const row = await tx<DbRefreshStateRow>('refresh_state')
|
||||
.select('location_key')
|
||||
.where('entity_ref', entityRef)
|
||||
.first();
|
||||
|
||||
const conflictingKey = row?.location_key;
|
||||
|
||||
// If there's no existing key we can't have a conflict
|
||||
if (!conflictingKey) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
if (conflictingKey !== locationKey) {
|
||||
return conflictingKey;
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
+72
@@ -0,0 +1,72 @@
|
||||
/*
|
||||
* 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 { Entity, stringifyEntityRef } from '@backstage/catalog-model';
|
||||
import { Knex } from 'knex';
|
||||
import { DbRefreshStateRow } from '../../tables';
|
||||
import { v4 as uuid } from 'uuid';
|
||||
import type { Logger } from 'winston';
|
||||
import { isDatabaseConflictError } from '@backstage/backend-common';
|
||||
|
||||
/**
|
||||
* Attempts to insert a new refresh state row for the given entity, returning
|
||||
* true if successful and false if there was a conflict.
|
||||
*/
|
||||
export async function insertUnprocessedEntity(options: {
|
||||
tx: Knex.Transaction;
|
||||
entity: Entity;
|
||||
hash: string;
|
||||
locationKey?: string;
|
||||
logger: Logger;
|
||||
}): Promise<boolean> {
|
||||
const { tx, entity, hash, logger, locationKey } = options;
|
||||
|
||||
const entityRef = stringifyEntityRef(entity);
|
||||
const serializedEntity = JSON.stringify(entity);
|
||||
|
||||
try {
|
||||
let query = tx<DbRefreshStateRow>('refresh_state').insert({
|
||||
entity_id: uuid(),
|
||||
entity_ref: entityRef,
|
||||
unprocessed_entity: serializedEntity,
|
||||
unprocessed_hash: hash,
|
||||
errors: '',
|
||||
location_key: locationKey,
|
||||
next_update_at: tx.fn.now(),
|
||||
last_discovery_at: tx.fn.now(),
|
||||
});
|
||||
|
||||
// TODO(Rugvip): only tested towards MySQL, 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.includes('pg')) {
|
||||
query = query.onConflict('entity_ref').ignore() as any; // type here does not match runtime
|
||||
}
|
||||
|
||||
// Postgres gives as an object with rowCount, SQLite gives us an array
|
||||
const result: { rowCount?: number; length?: number } = await query;
|
||||
return result.rowCount === 1 || result.length === 1;
|
||||
} catch (error) {
|
||||
// SQLite, or MySQL reached this rather than the rowCount check above
|
||||
if (!isDatabaseConflictError(error)) {
|
||||
throw error;
|
||||
} else {
|
||||
logger.debug(`Unable to insert a new refresh state row, ${error}`);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
+60
@@ -0,0 +1,60 @@
|
||||
/*
|
||||
* 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 { Entity, stringifyEntityRef } from '@backstage/catalog-model';
|
||||
import { Knex } from 'knex';
|
||||
import { DbRefreshStateRow } from '../../tables';
|
||||
|
||||
/**
|
||||
* Attempts to update an existing refresh state row, returning true if it was
|
||||
* updated and false if there was no entity with a matching ref and location key.
|
||||
*
|
||||
* Updating the entity will also cause it to be scheduled for immediate processing.
|
||||
*/
|
||||
export async function updateUnprocessedEntity(options: {
|
||||
tx: Knex.Transaction;
|
||||
entity: Entity;
|
||||
hash: string;
|
||||
locationKey?: string;
|
||||
}): Promise<boolean> {
|
||||
const { tx, entity, hash, locationKey } = options;
|
||||
|
||||
const entityRef = stringifyEntityRef(entity);
|
||||
const serializedEntity = JSON.stringify(entity);
|
||||
|
||||
const refreshResult = await tx<DbRefreshStateRow>('refresh_state')
|
||||
.update({
|
||||
unprocessed_entity: serializedEntity,
|
||||
unprocessed_hash: hash,
|
||||
location_key: locationKey,
|
||||
last_discovery_at: tx.fn.now(),
|
||||
// We only get to this point if a processed entity actually had any changes, or
|
||||
// if an entity provider requested this mutation, meaning that we can safely
|
||||
// bump the deferred entities to the front of the queue for immediate processing.
|
||||
next_update_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');
|
||||
});
|
||||
|
||||
return refreshResult === 1;
|
||||
}
|
||||
@@ -109,17 +109,12 @@ export type ListParentsResult = {
|
||||
entityRefs: string[];
|
||||
};
|
||||
|
||||
/**
|
||||
* The database abstraction layer for Entity Processor interactions.
|
||||
*/
|
||||
export interface ProcessingDatabase {
|
||||
transaction<T>(fn: (tx: Transaction) => Promise<T>): Promise<T>;
|
||||
|
||||
/**
|
||||
* Add unprocessed entities to the front of the processing queue using a mutation.
|
||||
*/
|
||||
replaceUnprocessedEntities(
|
||||
txOpaque: Transaction,
|
||||
options: ReplaceUnprocessedEntitiesOptions,
|
||||
): Promise<void>;
|
||||
|
||||
getProcessableEntities(
|
||||
txOpaque: Transaction,
|
||||
request: { processBatchSize: number },
|
||||
@@ -152,10 +147,25 @@ export interface ProcessingDatabase {
|
||||
options: UpdateProcessedEntityErrorsOptions,
|
||||
): Promise<void>;
|
||||
|
||||
listParents(
|
||||
txOpaque: Transaction,
|
||||
options: ListParentsOptions,
|
||||
): Promise<ListParentsResult>;
|
||||
}
|
||||
|
||||
/**
|
||||
* The database abstraction layer for Entity Provider interactions.
|
||||
*/
|
||||
export interface ProviderDatabase {
|
||||
transaction<T>(fn: (tx: Transaction) => Promise<T>): Promise<T>;
|
||||
|
||||
/**
|
||||
* Schedules a refresh of a given entityRef.
|
||||
* Add unprocessed entities to the front of the processing queue using a mutation.
|
||||
*/
|
||||
refresh(txOpaque: Transaction, options: RefreshOptions): Promise<void>;
|
||||
replaceUnprocessedEntities(
|
||||
txOpaque: Transaction,
|
||||
options: ReplaceUnprocessedEntitiesOptions,
|
||||
): Promise<void>;
|
||||
|
||||
/**
|
||||
* Schedules a refresh for every entity that has a matching set of refresh key stored for it.
|
||||
@@ -164,6 +174,14 @@ export interface ProcessingDatabase {
|
||||
txOpaque: Transaction,
|
||||
options: RefreshByKeyOptions,
|
||||
): Promise<void>;
|
||||
}
|
||||
|
||||
// TODO(Rugvip): This is only partial for now
|
||||
/**
|
||||
* The database abstraction layer for catalog access.
|
||||
*/
|
||||
export interface CatalogDatabase {
|
||||
transaction<T>(fn: (tx: Transaction) => Promise<T>): Promise<T>;
|
||||
|
||||
/**
|
||||
* Lists all ancestors of a given entityRef.
|
||||
@@ -175,8 +193,8 @@ export interface ProcessingDatabase {
|
||||
options: ListAncestorsOptions,
|
||||
): Promise<ListAncestorsResult>;
|
||||
|
||||
listParents(
|
||||
txOpaque: Transaction,
|
||||
options: ListParentsOptions,
|
||||
): Promise<ListParentsResult>;
|
||||
/**
|
||||
* Schedules a refresh of a given entityRef.
|
||||
*/
|
||||
refresh(txOpaque: Transaction, options: RefreshOptions): Promise<void>;
|
||||
}
|
||||
|
||||
@@ -29,6 +29,7 @@ import {
|
||||
import { defaultEntityDataParser } from './modules/util/parse';
|
||||
import { DefaultCatalogProcessingOrchestrator } from './processing/DefaultCatalogProcessingOrchestrator';
|
||||
import { applyDatabaseMigrations } from './database/migrations';
|
||||
import { DefaultCatalogDatabase } from './database/DefaultCatalogDatabase';
|
||||
import { DefaultProcessingDatabase } from './database/DefaultProcessingDatabase';
|
||||
import { ScmIntegrations } from '@backstage/integration';
|
||||
import { DefaultCatalogRulesEnforcer } from './ingestion/CatalogRules';
|
||||
@@ -50,6 +51,7 @@ import {
|
||||
processingResult,
|
||||
} from '@backstage/plugin-catalog-node';
|
||||
import { RefreshStateItem } from './database/types';
|
||||
import { DefaultProviderDatabase } from './database/DefaultProviderDatabase';
|
||||
|
||||
const voidLogger = getVoidLogger();
|
||||
|
||||
@@ -213,6 +215,14 @@ class TestHarness {
|
||||
|
||||
await applyDatabaseMigrations(db);
|
||||
|
||||
const catalogDatabase = new DefaultCatalogDatabase({
|
||||
database: db,
|
||||
logger,
|
||||
});
|
||||
const providerDatabase = new DefaultProviderDatabase({
|
||||
database: db,
|
||||
logger,
|
||||
});
|
||||
const processingDatabase = new DefaultProcessingDatabase({
|
||||
database: db,
|
||||
logger,
|
||||
@@ -272,11 +282,11 @@ class TestHarness {
|
||||
proxyProgressTracker,
|
||||
);
|
||||
|
||||
const refresh = new DefaultRefreshService({ database: processingDatabase });
|
||||
const refresh = new DefaultRefreshService({ database: catalogDatabase });
|
||||
|
||||
const provider = new TestProvider();
|
||||
|
||||
await connectEntityProviders(processingDatabase, [provider]);
|
||||
await connectEntityProviders(providerDatabase, [provider]);
|
||||
|
||||
return new TestHarness(
|
||||
catalog,
|
||||
|
||||
@@ -19,7 +19,7 @@ import {
|
||||
entityEnvelopeSchemaValidator,
|
||||
stringifyEntityRef,
|
||||
} from '@backstage/catalog-model';
|
||||
import { ProcessingDatabase } from '../database/types';
|
||||
import { ProviderDatabase } from '../database/types';
|
||||
import {
|
||||
EntityProvider,
|
||||
EntityProviderConnection,
|
||||
@@ -33,12 +33,12 @@ class Connection implements EntityProviderConnection {
|
||||
constructor(
|
||||
private readonly config: {
|
||||
id: string;
|
||||
processingDatabase: ProcessingDatabase;
|
||||
providerDatabase: ProviderDatabase;
|
||||
},
|
||||
) {}
|
||||
|
||||
async applyMutation(mutation: EntityProviderMutation): Promise<void> {
|
||||
const db = this.config.processingDatabase;
|
||||
const db = this.config.providerDatabase;
|
||||
|
||||
if (mutation.type === 'full') {
|
||||
this.check(mutation.entities.map(e => e.entity));
|
||||
@@ -75,7 +75,7 @@ class Connection implements EntityProviderConnection {
|
||||
}
|
||||
|
||||
async refresh(options: EntityProviderRefreshOptions): Promise<void> {
|
||||
const db = this.config.processingDatabase;
|
||||
const db = this.config.providerDatabase;
|
||||
|
||||
await db.transaction(async (tx: any) => {
|
||||
return db.refreshByRefreshKeys(tx, {
|
||||
@@ -96,14 +96,14 @@ class Connection implements EntityProviderConnection {
|
||||
}
|
||||
|
||||
export async function connectEntityProviders(
|
||||
db: ProcessingDatabase,
|
||||
db: ProviderDatabase,
|
||||
providers: EntityProvider[],
|
||||
) {
|
||||
await Promise.all(
|
||||
providers.map(async provider => {
|
||||
const connection = new Connection({
|
||||
id: provider.getProviderName(),
|
||||
processingDatabase: db,
|
||||
providerDatabase: db,
|
||||
});
|
||||
return provider.connect(connection);
|
||||
}),
|
||||
|
||||
@@ -29,6 +29,7 @@ export type EntityProcessingRequest = {
|
||||
entity: Entity;
|
||||
state?: JsonObject; // Versions for multiple deployments etc
|
||||
};
|
||||
|
||||
/**
|
||||
* The result of processing an entity.
|
||||
* @internal
|
||||
@@ -57,13 +58,18 @@ export type RefreshKeyData = {
|
||||
|
||||
/**
|
||||
* Responsible for executing the individual processing steps in order to fully process an entity.
|
||||
* @public
|
||||
*/
|
||||
export interface CatalogProcessingOrchestrator {
|
||||
process(request: EntityProcessingRequest): Promise<EntityProcessingResult>;
|
||||
}
|
||||
|
||||
/** @public */
|
||||
/**
|
||||
* Represents the engine that drives the processing loops. Some backend
|
||||
* instances may choose to not call start, if they focus only on API
|
||||
* interactions.
|
||||
*
|
||||
* @public
|
||||
*/
|
||||
export interface CatalogProcessingEngine {
|
||||
start(): Promise<void>;
|
||||
stop(): Promise<void>;
|
||||
|
||||
@@ -28,6 +28,8 @@ describe('AuthorizedEntitiesCatalog', () => {
|
||||
removeEntityByUid: jest.fn(),
|
||||
entityAncestry: jest.fn(),
|
||||
facets: jest.fn(),
|
||||
refresh: jest.fn(),
|
||||
listAncestors: jest.fn(),
|
||||
};
|
||||
const fakePermissionApi = {
|
||||
authorize: jest.fn(),
|
||||
|
||||
@@ -96,6 +96,8 @@ import {
|
||||
RESOURCE_TYPE_CATALOG_ENTITY,
|
||||
} from '@backstage/plugin-catalog-common';
|
||||
import { AuthorizedLocationService } from './AuthorizedLocationService';
|
||||
import { DefaultProviderDatabase } from '../database/DefaultProviderDatabase';
|
||||
import { DefaultCatalogDatabase } from '../database/DefaultCatalogDatabase';
|
||||
|
||||
/** @public */
|
||||
export type CatalogEnvironment = {
|
||||
@@ -431,6 +433,14 @@ export class CatalogBuilder {
|
||||
logger,
|
||||
refreshInterval: this.processingInterval,
|
||||
});
|
||||
const providerDatabase = new DefaultProviderDatabase({
|
||||
database: dbClient,
|
||||
logger,
|
||||
});
|
||||
const catalogDatabase = new DefaultCatalogDatabase({
|
||||
database: dbClient,
|
||||
logger,
|
||||
});
|
||||
const integrations = ScmIntegrations.fromConfig(config);
|
||||
const rulesEnforcer = DefaultCatalogRulesEnforcer.fromConfig(config);
|
||||
const orchestrator = new DefaultCatalogProcessingOrchestrator({
|
||||
@@ -520,7 +530,7 @@ export class CatalogBuilder {
|
||||
permissionEvaluator,
|
||||
);
|
||||
const refreshService = new AuthorizedRefreshService(
|
||||
new DefaultRefreshService({ database: processingDatabase }),
|
||||
new DefaultRefreshService({ database: catalogDatabase }),
|
||||
permissionEvaluator,
|
||||
);
|
||||
const router = await createRouter({
|
||||
@@ -534,7 +544,7 @@ export class CatalogBuilder {
|
||||
permissionIntegrationRouter,
|
||||
});
|
||||
|
||||
await connectEntityProviders(processingDatabase, entityProviders);
|
||||
await connectEntityProviders(providerDatabase, entityProviders);
|
||||
|
||||
return {
|
||||
processingEngine,
|
||||
|
||||
@@ -16,11 +16,14 @@
|
||||
|
||||
import { getVoidLogger } from '@backstage/backend-common';
|
||||
import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils';
|
||||
import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
|
||||
import { createHash } from 'crypto';
|
||||
import { Knex } from 'knex';
|
||||
import { v4 as uuid } from 'uuid';
|
||||
import { Logger } from 'winston';
|
||||
import { applyDatabaseMigrations } from '../database/migrations';
|
||||
import { DefaultCatalogDatabase } from '../database/DefaultCatalogDatabase';
|
||||
import { DefaultProcessingDatabase } from '../database/DefaultProcessingDatabase';
|
||||
import { applyDatabaseMigrations } from '../database/migrations';
|
||||
import {
|
||||
DbRefreshStateReferencesRow,
|
||||
DbRefreshStateRow,
|
||||
@@ -29,11 +32,9 @@ import { ProcessingDatabase } from '../database/types';
|
||||
import { DefaultCatalogProcessingEngine } from '../processing/DefaultCatalogProcessingEngine';
|
||||
import { EntityProcessingRequest } from '../processing/types';
|
||||
import { Stitcher } from '../stitching/Stitcher';
|
||||
import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
|
||||
import { v4 as uuid } from 'uuid';
|
||||
import { DefaultRefreshService } from './DefaultRefreshService';
|
||||
|
||||
describe('Refresh integration', () => {
|
||||
describe('DefaultRefreshService', () => {
|
||||
const defaultLogger = getVoidLogger();
|
||||
const databases = TestDatabases.create({
|
||||
ids: ['MYSQL_8', 'POSTGRES_13', 'POSTGRES_9', 'SQLITE_3'],
|
||||
@@ -47,11 +48,15 @@ describe('Refresh integration', () => {
|
||||
await applyDatabaseMigrations(knex);
|
||||
return {
|
||||
knex,
|
||||
db: new DefaultProcessingDatabase({
|
||||
processingDb: new DefaultProcessingDatabase({
|
||||
database: knex,
|
||||
logger,
|
||||
refreshInterval: () => 100,
|
||||
}),
|
||||
catalogDb: new DefaultCatalogDatabase({
|
||||
database: knex,
|
||||
logger,
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -176,10 +181,12 @@ describe('Refresh integration', () => {
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should refresh the parent location, %p',
|
||||
async databaseId => {
|
||||
const { knex, db } = await createDatabase(databaseId);
|
||||
const refreshService = new DefaultRefreshService({ database: db });
|
||||
const { knex, processingDb, catalogDb } = await createDatabase(
|
||||
databaseId,
|
||||
);
|
||||
const refreshService = new DefaultRefreshService({ database: catalogDb });
|
||||
const engine = await createPopulatedEngine({
|
||||
db,
|
||||
db: processingDb,
|
||||
knex,
|
||||
entities: [
|
||||
{
|
||||
@@ -220,10 +227,12 @@ describe('Refresh integration', () => {
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should refresh the location further up the tree, %p',
|
||||
async databaseId => {
|
||||
const { knex, db } = await createDatabase(databaseId);
|
||||
const refreshService = new DefaultRefreshService({ database: db });
|
||||
const { knex, processingDb, catalogDb } = await createDatabase(
|
||||
databaseId,
|
||||
);
|
||||
const refreshService = new DefaultRefreshService({ database: catalogDb });
|
||||
const engine = await createPopulatedEngine({
|
||||
db,
|
||||
db: processingDb,
|
||||
knex,
|
||||
entities: [
|
||||
{
|
||||
@@ -273,10 +282,12 @@ describe('Refresh integration', () => {
|
||||
'should refresh even when parent has no changes',
|
||||
async databaseId => {
|
||||
let secondRound = false;
|
||||
const { knex, db } = await createDatabase(databaseId);
|
||||
const refreshService = new DefaultRefreshService({ database: db });
|
||||
const { knex, processingDb, catalogDb } = await createDatabase(
|
||||
databaseId,
|
||||
);
|
||||
const refreshService = new DefaultRefreshService({ database: catalogDb });
|
||||
const engine = await createPopulatedEngine({
|
||||
db,
|
||||
db: processingDb,
|
||||
knex,
|
||||
entities: [
|
||||
{
|
||||
|
||||
@@ -14,13 +14,13 @@
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
import { DefaultProcessingDatabase } from '../database/DefaultProcessingDatabase';
|
||||
import { DefaultCatalogDatabase } from '../database/DefaultCatalogDatabase';
|
||||
import { RefreshOptions, RefreshService } from './types';
|
||||
|
||||
export class DefaultRefreshService implements RefreshService {
|
||||
private database: DefaultProcessingDatabase;
|
||||
private database: DefaultCatalogDatabase;
|
||||
|
||||
constructor(options: { database: DefaultProcessingDatabase }) {
|
||||
constructor(options: { database: DefaultCatalogDatabase }) {
|
||||
this.database = options.database;
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user