From ec43f706c71f728dfd57dfec20c65b004a71dbf6 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Mon, 19 Apr 2021 10:53:57 +0200 Subject: [PATCH] Copy migrations to migrationsv2 before migration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Fredrik Adelöw Co-authored-by: Patrik Oldsberg Signed-off-by: Johan Haals --- packages/backend/src/plugins/catalog.ts | 6 +- .../20210302150147_refresh_state.js | 92 +++++++++--------- plugins/catalog-backend/package.json | 1 + .../src/next/NextCatalogBuilder.ts | 28 +++++- .../src/next/NextEntitiesCatalog.ts | 61 ++++++++++++ plugins/catalog-backend/src/next/Stitcher.ts | 96 ++++++++++++++++++- .../next/database/ProcessingDatabaseImpl.ts | 11 --- tsconfig.json | 3 +- 8 files changed, 231 insertions(+), 67 deletions(-) rename plugins/catalog-backend/{migrations => migrationsv2}/20210302150147_refresh_state.js (98%) create mode 100644 plugins/catalog-backend/src/next/NextEntitiesCatalog.ts diff --git a/packages/backend/src/plugins/catalog.ts b/packages/backend/src/plugins/catalog.ts index 99ba769736..e58d8cd3eb 100644 --- a/packages/backend/src/plugins/catalog.ts +++ b/packages/backend/src/plugins/catalog.ts @@ -27,7 +27,11 @@ import { PluginEnvironment } from '../types'; export default async function createPlugin( env: PluginEnvironment, ): Promise { - // HIGHLY experimental rework of the software catalog + /* + * ** WARNING ** + * DO NOT enable the experimental catalog, it will brick your database migrations. + * This is solely for internal backstage development. + */ if (process.env.EXPERIMENTAL_CATALOG === '1') { const builder = new NextCatalogBuilder(env); const { diff --git a/plugins/catalog-backend/migrations/20210302150147_refresh_state.js b/plugins/catalog-backend/migrationsv2/20210302150147_refresh_state.js similarity index 98% rename from plugins/catalog-backend/migrations/20210302150147_refresh_state.js rename to plugins/catalog-backend/migrationsv2/20210302150147_refresh_state.js index a09d892146..56c594ddf2 100644 --- a/plugins/catalog-backend/migrations/20210302150147_refresh_state.js +++ b/plugins/catalog-backend/migrationsv2/20210302150147_refresh_state.js @@ -20,52 +20,6 @@ * @param {import('knex').Knex} knex */ exports.up = async function up(knex) { - await knex.schema.createTable('relations', table => { - table.comment('All relations between entities in the catalog'); - table - .text('originating_entity_id') - .references('entity_id') - .inTable('refresh_state') - .onDelete('CASCADE') - .notNullable() - .comment('The entity that provided the relation'); - table - .text('source_entity_ref') - .notNullable() - .comment('The entity reference of the source entity of the relation'); - table - .text('type') - .notNullable() - .comment('The type of the relation between the entities'); - table - .text('target_entity_ref') - .notNullable() - .comment('The entity reference of the target entity of the relation'); - - table.index( - ['source_entity_ref', 'type', 'target_entity_ref'], - 'relations_full_ref_idx', - ); - }); - - await knex.schema.createTable('final_entities', table => { - table.comment( - 'This table contains the final entity result after processing and stitching', - ); - table - .text('entity_id') - .primary() - .notNullable() - .references('entity_id') - .inTable('refresh_state') - .onDelete('CASCADE') - .comment( - 'Entity ID which correspond to the ID in the refresh_state table', - ); - - table.text('finalized_entity').notNullable().comment('The final entity'); - }); - await knex.schema.createTable('refresh_state', table => { table.comment( 'Location refresh states. Every individual location (that was ever directly or indirectly discovered) and entity has an entry in this table. It therefore represents the entire live set of things that the refresh loop considers.', @@ -116,6 +70,24 @@ exports.up = async function up(knex) { table.index('next_update_at', 'refresh_state_next_update_at_idx'); }); + await knex.schema.createTable('final_entities', table => { + table.comment( + 'This table contains the final entity result after processing and stitching', + ); + table + .text('entity_id') + .primary() + .notNullable() + .references('entity_id') + .inTable('refresh_state') + .onDelete('CASCADE') + .comment( + 'Entity ID which correspond to the ID in the refresh_state table', + ); + table.text('etag').notNullable().comment('Etag to be used for caching'); + table.text('finalized_entity').notNullable().comment('The final entity'); + }); + await knex.schema.createTable('refresh_state_references', table => { table.comment( 'Holds edges between refresh state rows. Every time when an entity is processed and emits another entity, an edge will be stored to represent that fact. This is used to detect orphans and ultimately deletions.', @@ -155,6 +127,34 @@ exports.up = async function up(knex) { 'refresh_state_references_target_entity_id_idx', ); }); + + await knex.schema.createTable('relations', table => { + table.comment('All relations between entities in the catalog'); + table + .text('originating_entity_id') + .references('entity_id') + .inTable('refresh_state') + .onDelete('CASCADE') + .notNullable() + .comment('The entity that provided the relation'); + table + .text('source_entity_ref') + .notNullable() + .comment('The entity reference of the source entity of the relation'); + table + .text('type') + .notNullable() + .comment('The type of the relation between the entities'); + table + .text('target_entity_ref') + .notNullable() + .comment('The entity reference of the target entity of the relation'); + + table.index( + ['source_entity_ref', 'type', 'target_entity_ref'], + 'relations_full_ref_idx', + ); + }); }; /** diff --git a/plugins/catalog-backend/package.json b/plugins/catalog-backend/package.json index 6011562078..64879df777 100644 --- a/plugins/catalog-backend/package.json +++ b/plugins/catalog-backend/package.json @@ -80,6 +80,7 @@ "files": [ "dist", "migrations/**/*.{js,d.ts}", + "migrationsv2/**/*.{js,d.ts}", "config.d.ts" ], "configSchema": "config.d.ts" diff --git a/plugins/catalog-backend/src/next/NextCatalogBuilder.ts b/plugins/catalog-backend/src/next/NextCatalogBuilder.ts index 95c767407e..b7e860cce2 100644 --- a/plugins/catalog-backend/src/next/NextCatalogBuilder.ts +++ b/plugins/catalog-backend/src/next/NextCatalogBuilder.ts @@ -14,7 +14,12 @@ * limitations under the License. */ -import { PluginDatabaseManager, UrlReader } from '@backstage/backend-common'; +import { + PluginDatabaseManager, + resolvePackagePath, + UrlReader, +} from '@backstage/backend-common'; +import fs from 'fs-extra'; import { DefaultNamespaceEntityPolicy, EntityPolicies, @@ -30,12 +35,10 @@ import { ScmIntegrations } from '@backstage/integration'; import lodash from 'lodash'; import { Logger } from 'winston'; import { - DatabaseEntitiesCatalog, DatabaseLocationsCatalog, EntitiesCatalog, LocationsCatalog, } from '../catalog'; -import { DatabaseManager } from '../database'; import { AnnotateLocationEntityProcessor, BitbucketDiscoveryProcessor, @@ -68,7 +71,9 @@ import { DatabaseLocationProvider } from '../next/DatabaseLocationProvider'; import { LocationStoreImpl } from '../next/LocationStoreImpl'; import { ProcessingStateManagerImpl } from '../next/ProcessingStateManagerImpl'; import { CatalogProcessingEngine } from '../next/types'; +import { NextEntitiesCatalog } from './NextEntitiesCatalog'; import { Stitcher } from './Stitcher'; +import { CommonDatabase } from '../database/CommonDatabase'; export type CatalogEnvironment = { logger: Logger; @@ -239,7 +244,20 @@ export class NextCatalogBuilder { const parser = this.parser || defaultEntityDataParser; const dbClient = await database.getClient(); - const db = await DatabaseManager.createDatabase(dbClient, { logger }); + const allMigrations = resolvePackagePath( + '@backstage/plugin-catalog-backend', + 'migrations', + ); + + const migrationsDir = resolvePackagePath( + '@backstage/plugin-catalog-backend', + 'migrationsv2', + ); + await fs.copy(allMigrations, migrationsDir); + await dbClient.migrate.latest({ + directory: migrationsDir, + }); + const db = new CommonDatabase(dbClient, logger); const processingDatabase = new ProcessingDatabaseImpl(dbClient, logger); const stateManager = new ProcessingStateManagerImpl(processingDatabase); @@ -251,7 +269,7 @@ export class NextCatalogBuilder { parser, policy, }); - const entitiesCatalog = new DatabaseEntitiesCatalog(db, logger); + const entitiesCatalog = new NextEntitiesCatalog(dbClient, logger); const locationStore = new LocationStoreImpl(db); const dbLocationProvider = new DatabaseLocationProvider(locationStore); diff --git a/plugins/catalog-backend/src/next/NextEntitiesCatalog.ts b/plugins/catalog-backend/src/next/NextEntitiesCatalog.ts new file mode 100644 index 0000000000..39d3bf2a63 --- /dev/null +++ b/plugins/catalog-backend/src/next/NextEntitiesCatalog.ts @@ -0,0 +1,61 @@ +/* + * Copyright 2020 Spotify AB + * + * 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 { Logger } from 'winston'; +import { Knex } from 'knex'; +import { DbFinalEntitiesRow } from './Stitcher'; +import { EntitiesCatalog } from '../catalog'; +import { EntitiesRequest, EntitiesResponse } from '../catalog/types'; + +export class NextEntitiesCatalog implements EntitiesCatalog { + constructor( + private readonly database: Knex, + private readonly logger: Logger, + ) {} + + async entities(request?: EntitiesRequest): Promise { + if (request?.fields) { + throw new Error('Fields extraction is not implemented'); + } + if (request?.pagination) { + throw new Error('Pagination is not implemented'); + } + if (request?.filter) { + throw new Error('Filters are not implemented'); + } + + const dbResponse = await this.database( + 'final_entities', + ).select(); + + const entities = dbResponse.map(e => JSON.parse(e.finalized_entity)); + + return { + entities, + pageInfo: { + hasNextPage: false, + }, + }; + } + + async removeEntityByUid(_uid: string): Promise { + throw new Error('Not implemented'); + } + + async batchAddOrUpdateEntities(): Promise { + throw new Error('Not implemented'); + } +} diff --git a/plugins/catalog-backend/src/next/Stitcher.ts b/plugins/catalog-backend/src/next/Stitcher.ts index 8adc93bfad..29e7363b80 100644 --- a/plugins/catalog-backend/src/next/Stitcher.ts +++ b/plugins/catalog-backend/src/next/Stitcher.ts @@ -1,5 +1,3 @@ -import { Transaction } from '../database'; - /* * Copyright 2021 Spotify AB * @@ -18,7 +16,28 @@ import { Transaction } from '../database'; import { Knex } from 'knex'; import { Logger } from 'winston'; +import { Transaction } from '../database'; import { ConflictError } from '@backstage/errors'; +import { + DbRefreshStateReferences, + DbRefreshStateRow, + DbRelationsRow, +} from './database/ProcessingDatabaseImpl'; +import { Entity, parseEntityRef } from '@backstage/catalog-model'; +import { createHash } from 'crypto'; +import stableStringify from 'fast-json-stable-stringify'; + +export type DbFinalEntitiesRow = { + entity_id: string; + etag: string; + finalized_entity: string; +}; + +function generateEntityEtag(entity: Entity) { + return createHash('sha1') + .update(stableStringify({ ...entity })) + .digest('hex'); +} export class Stitcher { constructor( @@ -27,7 +46,74 @@ export class Stitcher { ) {} async stitch(entityRefs: Set) { - console.log(entityRefs); + for (const entityRef of entityRefs) { + await this.transaction(async txOpaque => { + const tx = txOpaque as Knex.Transaction; + const [result] = await tx('refresh_state') + .select('entity_id', 'processed_entity') + .where({ entity_ref: entityRef }); + + if (!result) { + this.logger.debug( + `Unable to stitch ${entityRef}, item does not exist in refresh state table`, + ); + return; + } else if (!result.processed_entity) { + this.logger.debug( + `Unable to stitch ${entityRef}, the entity has not yet been processed`, + ); + return; + } + + const entity: Entity = JSON.parse(result.processed_entity); + + const entityId = entity?.metadata?.uid; + if (!entityId) { + this.logger.error(`missing ID in entity ${JSON.stringify(entity)}`); + return; + } + + const [reference_count_result] = await tx( + 'refresh_state_references', + ) + .where({ target_entity_id: entity.metadata.uid }) + .count({ reference_count: 'target_entity_id' }); + console.log( + 'DEBUG: reference_count_result =', + reference_count_result, + entityId, + ); + if (Number(reference_count_result.reference_count) === 0) { + this.logger.debug(`${entityRef} is orphan`); + entity.metadata.annotations = { + ...entity.metadata.annotations, + ['backstage.io/orphan']: 'true', + }; + } + + const relationResults = await tx('relations') + .where({ originating_entity_id: entityId }) + .select(); + + // TODO: entityRef is lower case and should be uppercase in the final result. + entity.relations = relationResults.map(relation => ({ + type: relation.type, + target: parseEntityRef(relation.target_entity_ref), + })); + entity.metadata.generation = 1; + const etag = generateEntityEtag(entity); + entity.metadata.etag = etag; + console.log(JSON.stringify(entity, null, 2)); + await tx('final_entities') + .insert({ + finalized_entity: JSON.stringify(entity), + entity_id: entityId, + etag, + }) + .onConflict('entity_id') + .merge(['finalized_entity', 'etag']); + }); + } } private async transaction( @@ -45,6 +131,10 @@ export class Stitcher { { // If we explicitly trigger a rollback, don't fail. doNotRejectOnRollback: true, + isolationLevel: + this.database.client.config.client === 'sqlite3' + ? undefined // sqlite3 only supports serializable transactions, ignoring the isolation level param + : 'serializable', }, ); diff --git a/plugins/catalog-backend/src/next/database/ProcessingDatabaseImpl.ts b/plugins/catalog-backend/src/next/database/ProcessingDatabaseImpl.ts index f479c9a4f8..7c56fcd259 100644 --- a/plugins/catalog-backend/src/next/database/ProcessingDatabaseImpl.ts +++ b/plugins/catalog-backend/src/next/database/ProcessingDatabaseImpl.ts @@ -27,8 +27,6 @@ import { } from './types'; import type { Logger } from 'winston'; import { Entity, stringifyEntityRef } from '@backstage/catalog-model'; -import { createHash } from 'crypto'; -import stableStringify from 'fast-json-stable-stringify'; import { v4 as uuid } from 'uuid'; export type DbRefreshStateRow = { @@ -55,12 +53,6 @@ export type DbRefreshStateReferences = { target_entity_id: string; }; -function generateEntityEtag(entity: Entity) { - return createHash('sha1') - .update(stableStringify({ ...entity })) - .digest('hex'); -} - // 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 @@ -108,9 +100,6 @@ export class ProcessingDatabaseImpl implements ProcessingDatabase { // Update fragments - // Update relations - const entityRef = stringifyEntityRef(processedEntity); - // Delete old relations await tx('relations') .where({ originating_entity_id: id }) diff --git a/tsconfig.json b/tsconfig.json index d6cb6f4c96..6139c4a7f0 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -4,7 +4,8 @@ "packages/*/src", "plugins/*/src", "plugins/*/dev", - "plugins/*/migrations" + "plugins/*/migrations", + "plugins/catalog-backend/migrationsv2" ], "compilerOptions": { "outDir": "dist-types",