From effc37cf492deb7bb92c4135ff56754161927644 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Fri, 9 Dec 2022 14:29:45 +0100 Subject: [PATCH] catalog-backend: split out some common DB operations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Fredrik Adelöw Signed-off-by: Patrik Oldsberg --- .../src/database/DefaultProcessingDatabase.ts | 151 +++--------------- .../src/database/DefaultProviderDatabase.ts | 134 +--------------- .../refreshState/checkLocationKeyConflict.ts | 47 ++++++ .../refreshState/insertUnprocessedEntity.ts | 70 ++++++++ .../refreshState/updateUnprocessedEntity.ts | 58 +++++++ 5 files changed, 201 insertions(+), 259 deletions(-) create mode 100644 plugins/catalog-backend/src/database/operations/refreshState/checkLocationKeyConflict.ts create mode 100644 plugins/catalog-backend/src/database/operations/refreshState/insertUnprocessedEntity.ts create mode 100644 plugins/catalog-backend/src/database/operations/refreshState/updateUnprocessedEntity.ts diff --git a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts index c32f7d6f3d..6e77a6a0d8 100644 --- a/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/database/DefaultProcessingDatabase.ts @@ -18,18 +18,7 @@ import { Entity, stringifyEntityRef } from '@backstage/catalog-model'; 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, - UpdateProcessedEntityOptions, - UpdateEntityCacheOptions, - ListParentsOptions, - ListParentsResult, -} from './types'; import { ProcessingIntervalFunction } from '../processing/refresh'; import { rethrowError, timestampToDateTime } from './conversion'; import { initDatabaseMetrics } from './metrics'; @@ -39,10 +28,22 @@ 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 @@ -291,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, @@ -438,7 +322,7 @@ 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, @@ -449,10 +333,11 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { continue; } - const inserted = await this.insertUnprocessedEntity( + const inserted = await insertUnprocessedEntity( tx, entity, hash, + this.options.logger, locationKey, ); if (inserted) { @@ -463,7 +348,7 @@ 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, diff --git a/plugins/catalog-backend/src/database/DefaultProviderDatabase.ts b/plugins/catalog-backend/src/database/DefaultProviderDatabase.ts index 05a1011c8d..9dbdc5f516 100644 --- a/plugins/catalog-backend/src/database/DefaultProviderDatabase.ts +++ b/plugins/catalog-backend/src/database/DefaultProviderDatabase.ts @@ -15,13 +15,16 @@ */ import { isDatabaseConflictError } from '@backstage/backend-common'; -import { Entity, stringifyEntityRef } from '@backstage/catalog-model'; +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 { checkLocationKeyConflict } from './operations/refreshState/checkLocationKeyConflict'; +import { insertUnprocessedEntity } from './operations/refreshState/insertUnprocessedEntity'; +import { updateUnprocessedEntity } from './operations/refreshState/updateUnprocessedEntity'; import { DbRefreshStateReferencesRow, DbRefreshStateRow } from './tables'; import { ProviderDatabase, @@ -255,17 +258,13 @@ export class DefaultProviderDatabase implements ProviderDatabase { const entityRef = stringifyEntityRef(entity); try { - let ok = await this.updateUnprocessedEntity( - tx, - entity, - hash, - locationKey, - ); + let ok = await updateUnprocessedEntity(tx, entity, hash, locationKey); if (!ok) { - ok = await this.insertUnprocessedEntity( + ok = await insertUnprocessedEntity( tx, entity, hash, + this.options.logger, locationKey, ); } @@ -278,7 +277,7 @@ export class DefaultProviderDatabase implements ProviderDatabase { target_entity_ref: entityRef, }); } else { - const conflictingKey = await this.checkLocationKeyConflict( + const conflictingKey = await checkLocationKeyConflict( tx, entityRef, locationKey, @@ -317,123 +316,6 @@ export class DefaultProviderDatabase implements ProviderDatabase { .update({ next_update_at: tx.fn.now() }); } - /** - * 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 async createDelta( tx: Knex.Transaction, options: ReplaceUnprocessedEntitiesOptions, 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..a256b36729 --- /dev/null +++ b/plugins/catalog-backend/src/database/operations/refreshState/checkLocationKeyConflict.ts @@ -0,0 +1,47 @@ +/* + * 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( + 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; +} 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..bf34fd2ce8 --- /dev/null +++ b/plugins/catalog-backend/src/database/operations/refreshState/insertUnprocessedEntity.ts @@ -0,0 +1,70 @@ +/* + * 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( + tx: Knex.Transaction, + entity: Entity, + hash: string, + logger: Logger, + 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 { + 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..084454de66 --- /dev/null +++ b/plugins/catalog-backend/src/database/operations/refreshState/updateUnprocessedEntity.ts @@ -0,0 +1,58 @@ +/* + * 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( + 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; +}