diff --git a/.changeset/lucky-chicken-greet.md b/.changeset/lucky-chicken-greet.md new file mode 100644 index 0000000000..7e061dc040 --- /dev/null +++ b/.changeset/lucky-chicken-greet.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend': patch +--- + +Internal code reorganization. diff --git a/plugins/catalog-backend/api-report.md b/plugins/catalog-backend/api-report.md index a46abda5d3..3e6c5ef676 100644 --- a/plugins/catalog-backend/api-report.md +++ b/plugins/catalog-backend/api-report.md @@ -239,7 +239,7 @@ export type CatalogPermissionRule< // @alpha export const catalogPlugin: (options?: undefined) => BackendFeature; -// @public (undocumented) +// @public export interface CatalogProcessingEngine { // (undocumented) start(): Promise; diff --git a/plugins/catalog-backend/src/database/DefaultCatalogDatabase.test.ts b/plugins/catalog-backend/src/database/DefaultCatalogDatabase.test.ts new file mode 100644 index 0000000000..b873f6a985 --- /dev/null +++ b/plugins/catalog-backend/src/database/DefaultCatalogDatabase.test.ts @@ -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('refresh_state').insert( + makeEntity('location:default/root-1'), + ); + await knex('refresh_state').insert( + makeEntity('location:default/root-2'), + ); + await knex('refresh_state').insert( + makeEntity('component:default/foobar'), + ); + + await knex( + 'refresh_state_references', + ).insert({ + source_key: 'source', + target_entity_ref: 'location:default/root-2', + }); + await knex( + 'refresh_state_references', + ).insert({ + source_entity_ref: 'location:default/root-2', + target_entity_ref: 'location:default/root-1', + }); + await knex( + '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', + ]); + }, + ); + }); +}); diff --git a/plugins/catalog-backend/src/database/DefaultCatalogDatabase.ts b/plugins/catalog-backend/src/database/DefaultCatalogDatabase.ts new file mode 100644 index 0000000000..98b100c91e --- /dev/null +++ b/plugins/catalog-backend/src/database/DefaultCatalogDatabase.ts @@ -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(fn: (tx: Transaction) => Promise): Promise { + 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 { + const tx = txOpaque as Knex.Transaction; + const { entityRef } = options; + const entityRefs = new Array(); + + let currentRef = entityRef.toLocaleLowerCase('en-US'); + for (let depth = 1; depth <= MAX_ANCESTOR_DEPTH; depth += 1) { + const rows = await tx( + '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 { + const tx = txOpaque as Knex.Transaction; + const { entityRef } = options; + + const updateResult = await tx('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`); + } + } +} diff --git a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.test.ts index 1056557706..4fbb38b4fc 100644 --- a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.test.ts +++ b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.test.ts @@ -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( - 'refresh_state', - ).select(); - - const currentRefRowState = await knex( - '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( - 'refresh_state', - ).select(); - - const currentRefRowState = await knex( - '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( - 'refresh_state', - ).select(); - - const currentRefRowState = await knex( - '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( - 'refresh_state', - ).select(); - - const currentRefRowState = await knex( - '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( - 'refresh_state', - ).select(); - - const currentRefRowState = await knex( - '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( - 'refresh_state', - ).select(); - expect(currentRefreshState).toEqual([ - expect.objectContaining({ - entity_ref: 'location:default/replaced', - location_key: 'file:///tmp/foobar', - }), - ]); - - const currentRefRowState = await knex( - 'refresh_state_references', - ).select(); - expect(currentRefRowState).toEqual([ - expect.objectContaining({ - source_key: 'lols', - target_entity_ref: 'location:default/replaced', - }), - ]); - }, - 60_000, - ); - - 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('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('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('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('refresh_state').insert( - makeEntity('location:default/root-1'), - ); - await knex('refresh_state').insert( - makeEntity('location:default/root-2'), - ); - await knex('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) { diff --git a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts index 97259bce29..1127deed7e 100644 --- a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts @@ -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 { - 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('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('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( - '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 { - const tx = txOpaque as Knex.Transaction; - const { entityRef } = options; - const entityRefs = new Array(); - - let currentRef = entityRef.toLocaleLowerCase('en-US'); - for (let depth = 1; depth <= MAX_ANCESTOR_DEPTH; depth += 1) { - const rows = await tx( - '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 { - const tx = txOpaque as Knex.Transaction; - const { entityRef } = options; - - const updateResult = await tx('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('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(fn: (tx: Transaction) => Promise): Promise { 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 { - const entityRef = stringifyEntityRef(entity); - const serializedEntity = JSON.stringify(entity); - - const refreshResult = await tx('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 { - const entityRef = stringifyEntityRef(entity); - const serializedEntity = JSON.stringify(entity); - - try { - let query = tx('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 { - const row = await tx('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( - 'refresh_state_references', - ) - .leftJoin('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}`, diff --git a/plugins/catalog-backend/src/database/DefaultProviderDatabase.test.ts b/plugins/catalog-backend/src/database/DefaultProviderDatabase.test.ts new file mode 100644 index 0000000000..43f2209b60 --- /dev/null +++ b/plugins/catalog-backend/src/database/DefaultProviderDatabase.test.ts @@ -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('refresh_state_references').insert( + ref, + ); + }; + + const insertRefreshStateRow = async (db: Knex, ref: DbRefreshStateRow) => { + await db('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( + 'refresh_state', + ).select(); + + const currentRefRowState = await knex( + '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( + 'refresh_state', + ).select(); + + const currentRefRowState = await knex( + '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( + 'refresh_state', + ).select(); + + const currentRefRowState = await knex( + '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( + 'refresh_state', + ).select(); + + const currentRefRowState = await knex( + '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( + 'refresh_state', + ).select(); + + const currentRefRowState = await knex( + '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( + 'refresh_state', + ).select(); + expect(currentRefreshState).toEqual([ + expect.objectContaining({ + entity_ref: 'location:default/replaced', + location_key: 'file:///tmp/foobar', + }), + ]); + + const currentRefRowState = await knex( + 'refresh_state_references', + ).select(); + expect(currentRefRowState).toEqual([ + expect.objectContaining({ + source_key: 'lols', + target_entity_ref: 'location:default/replaced', + }), + ]); + }, + 60_000, + ); + + 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('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('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('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, + ); + }); +}); diff --git a/plugins/catalog-backend/src/database/DefaultProviderDatabase.ts b/plugins/catalog-backend/src/database/DefaultProviderDatabase.ts new file mode 100644 index 0000000000..f1073fd984 --- /dev/null +++ b/plugins/catalog-backend/src/database/DefaultProviderDatabase.ts @@ -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(fn: (tx: Transaction) => Promise): Promise { + 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 { + 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( + '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( + 'refresh_state_references', + ) + .leftJoin('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 }; + } +} diff --git a/plugins/catalog-backend/src/database/operations/provider/deleteWithEagerPruningOfChildren.ts b/plugins/catalog-backend/src/database/operations/provider/deleteWithEagerPruningOfChildren.ts new file mode 100644 index 0000000000..3d642cbf3a --- /dev/null +++ b/plugins/catalog-backend/src/database/operations/provider/deleteWithEagerPruningOfChildren.ts @@ -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 { + 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('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('refresh_state_references') + .where('source_key', '=', sourceKey) + .whereIn('target_entity_ref', refs) + .delete(); + } + + return removedCount; +} diff --git a/plugins/catalog-backend/src/database/operations/provider/refreshByRefreshKeys.ts b/plugins/catalog-backend/src/database/operations/provider/refreshByRefreshKeys.ts new file mode 100644 index 0000000000..dc34c9cdf0 --- /dev/null +++ b/plugins/catalog-backend/src/database/operations/provider/refreshByRefreshKeys.ts @@ -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 { + const { tx, keys } = options; + + await tx('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() }); +} diff --git a/plugins/catalog-backend/src/database/operations/refreshState/checkLocationKeyConflict.ts b/plugins/catalog-backend/src/database/operations/refreshState/checkLocationKeyConflict.ts new file mode 100644 index 0000000000..97d376f9a4 --- /dev/null +++ b/plugins/catalog-backend/src/database/operations/refreshState/checkLocationKeyConflict.ts @@ -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 { + const { tx, entityRef, locationKey } = options; + + const row = await tx('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; +} diff --git a/plugins/catalog-backend/src/database/operations/refreshState/insertUnprocessedEntity.ts b/plugins/catalog-backend/src/database/operations/refreshState/insertUnprocessedEntity.ts new file mode 100644 index 0000000000..8e1c6edc87 --- /dev/null +++ b/plugins/catalog-backend/src/database/operations/refreshState/insertUnprocessedEntity.ts @@ -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 { + const { tx, entity, hash, logger, locationKey } = options; + + const entityRef = stringifyEntityRef(entity); + const serializedEntity = JSON.stringify(entity); + + try { + let query = tx('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; + } + } +} diff --git a/plugins/catalog-backend/src/database/operations/refreshState/updateUnprocessedEntity.ts b/plugins/catalog-backend/src/database/operations/refreshState/updateUnprocessedEntity.ts new file mode 100644 index 0000000000..e26db4429b --- /dev/null +++ b/plugins/catalog-backend/src/database/operations/refreshState/updateUnprocessedEntity.ts @@ -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 { + const { tx, entity, hash, locationKey } = options; + + const entityRef = stringifyEntityRef(entity); + const serializedEntity = JSON.stringify(entity); + + const refreshResult = await tx('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; +} diff --git a/plugins/catalog-backend/src/database/types.ts b/plugins/catalog-backend/src/database/types.ts index 345aee6ebd..56e5f48f14 100644 --- a/plugins/catalog-backend/src/database/types.ts +++ b/plugins/catalog-backend/src/database/types.ts @@ -109,17 +109,12 @@ export type ListParentsResult = { entityRefs: string[]; }; +/** + * The database abstraction layer for Entity Processor interactions. + */ export interface ProcessingDatabase { transaction(fn: (tx: Transaction) => Promise): Promise; - /** - * Add unprocessed entities to the front of the processing queue using a mutation. - */ - replaceUnprocessedEntities( - txOpaque: Transaction, - options: ReplaceUnprocessedEntitiesOptions, - ): Promise; - getProcessableEntities( txOpaque: Transaction, request: { processBatchSize: number }, @@ -152,10 +147,25 @@ export interface ProcessingDatabase { options: UpdateProcessedEntityErrorsOptions, ): Promise; + listParents( + txOpaque: Transaction, + options: ListParentsOptions, + ): Promise; +} + +/** + * The database abstraction layer for Entity Provider interactions. + */ +export interface ProviderDatabase { + transaction(fn: (tx: Transaction) => Promise): Promise; + /** - * 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; + replaceUnprocessedEntities( + txOpaque: Transaction, + options: ReplaceUnprocessedEntitiesOptions, + ): Promise; /** * 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; +} + +// TODO(Rugvip): This is only partial for now +/** + * The database abstraction layer for catalog access. + */ +export interface CatalogDatabase { + transaction(fn: (tx: Transaction) => Promise): Promise; /** * Lists all ancestors of a given entityRef. @@ -175,8 +193,8 @@ export interface ProcessingDatabase { options: ListAncestorsOptions, ): Promise; - listParents( - txOpaque: Transaction, - options: ListParentsOptions, - ): Promise; + /** + * Schedules a refresh of a given entityRef. + */ + refresh(txOpaque: Transaction, options: RefreshOptions): Promise; } diff --git a/plugins/catalog-backend/src/integration.test.ts b/plugins/catalog-backend/src/integration.test.ts index a63119bb67..4ba046834b 100644 --- a/plugins/catalog-backend/src/integration.test.ts +++ b/plugins/catalog-backend/src/integration.test.ts @@ -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, diff --git a/plugins/catalog-backend/src/processing/connectEntityProviders.ts b/plugins/catalog-backend/src/processing/connectEntityProviders.ts index 3eb638ce7a..30d30ca9b6 100644 --- a/plugins/catalog-backend/src/processing/connectEntityProviders.ts +++ b/plugins/catalog-backend/src/processing/connectEntityProviders.ts @@ -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 { - 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 { - 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); }), diff --git a/plugins/catalog-backend/src/processing/types.ts b/plugins/catalog-backend/src/processing/types.ts index ee5e5c48f2..c0d0e5db0d 100644 --- a/plugins/catalog-backend/src/processing/types.ts +++ b/plugins/catalog-backend/src/processing/types.ts @@ -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; } -/** @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; stop(): Promise; diff --git a/plugins/catalog-backend/src/service/AuthorizedEntitiesCatalog.test.ts b/plugins/catalog-backend/src/service/AuthorizedEntitiesCatalog.test.ts index cbaf876071..829e41c21e 100644 --- a/plugins/catalog-backend/src/service/AuthorizedEntitiesCatalog.test.ts +++ b/plugins/catalog-backend/src/service/AuthorizedEntitiesCatalog.test.ts @@ -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(), diff --git a/plugins/catalog-backend/src/service/CatalogBuilder.ts b/plugins/catalog-backend/src/service/CatalogBuilder.ts index 8aab74c4ba..48bf6a08ef 100644 --- a/plugins/catalog-backend/src/service/CatalogBuilder.ts +++ b/plugins/catalog-backend/src/service/CatalogBuilder.ts @@ -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, diff --git a/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts b/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts index 3ed4f3fe36..eafa0b0c14 100644 --- a/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts +++ b/plugins/catalog-backend/src/service/DefaultRefreshService.test.ts @@ -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: [ { diff --git a/plugins/catalog-backend/src/service/DefaultRefreshService.ts b/plugins/catalog-backend/src/service/DefaultRefreshService.ts index 3b982a0e46..99b78f900b 100644 --- a/plugins/catalog-backend/src/service/DefaultRefreshService.ts +++ b/plugins/catalog-backend/src/service/DefaultRefreshService.ts @@ -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; }