From 597bf5a3f2b90a3864a3e3f082be8069212fee47 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Wed, 26 May 2021 16:47:31 +0200 Subject: [PATCH 1/4] catalog/next: Replace stitch transactions with ticket MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Patrik Oldsberg Co-authored-by: Fredrik Adelöw Signed-off-by: Johan Haals --- .../20210302150147_refresh_state.js | 14 +- .../next/DefaultCatalogProcessingEngine.ts | 100 +++--- .../src/next/NextEntitiesCatalog.ts | 5 +- .../catalog-backend/src/next/Stitcher.test.ts | 288 +++++++++--------- plugins/catalog-backend/src/next/Stitcher.ts | 109 +++---- .../DefaultProcessingDatabase.test.ts | 5 + .../database/DefaultProcessingDatabase.ts | 10 + 7 files changed, 280 insertions(+), 251 deletions(-) diff --git a/plugins/catalog-backend/migrationsv2/20210302150147_refresh_state.js b/plugins/catalog-backend/migrationsv2/20210302150147_refresh_state.js index 4c03405910..4e6cf4f282 100644 --- a/plugins/catalog-backend/migrationsv2/20210302150147_refresh_state.js +++ b/plugins/catalog-backend/migrationsv2/20210302150147_refresh_state.js @@ -92,8 +92,14 @@ exports.up = async function up(knex) { 'Stable hash of the entity data, to be used for caching and avoiding redundant work', ); table - .text('final_entity') + .text('stitch_ticket') .notNullable() + .comment( + 'A random value representing a unique stitch attempt ticket, that gets updated each time that a stitching attempt is made on the entity', + ); + table + .text('final_entity') + .nullable() .comment('The JSON encoded final entity'); table.index('entity_id', 'final_entities_entity_id_idx'); }); @@ -192,9 +198,9 @@ exports.up = async function up(knex) { */ exports.down = async function down(knex) { await knex.schema.alterTable('refresh_state_references', table => { - table.dropIndex([], 'refresh_state_references_source_special_key_idx'); - table.dropIndex([], 'refresh_state_references_source_entity_id_idx'); - table.dropIndex([], 'refresh_state_references_target_entity_id_idx'); + table.dropIndex([], 'refresh_state_references_source_key_idx'); + table.dropIndex([], 'refresh_state_references_source_entity_ref_idx'); + table.dropIndex([], 'refresh_state_references_target_entity_ref_idx'); }); await knex.schema.alterTable('refresh_state', table => { table.dropUnique([], 'refresh_state_entity_ref_uniq'); diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index b7b6dfa297..e908eb95bc 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -110,56 +110,62 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { return; } - const { id, state, unprocessedEntity } = items[0]; - - const result = await this.orchestrator.process({ - entity: unprocessedEntity, - state, - }); - for (const error of result.errors) { - this.logger.warn(error.message); - } - const errorsString = JSON.stringify( - result.errors.map(e => serializeError(e)), - ); - - // If the result was marked as not OK, it signals that some part of the - // processing pipeline threw an exception. This can happen both as part of - // non-catastrophic things such as due to validation errors, as well as if - // something fatal happens inside the processing for other reasons. In any - // case, this means we can't trust that anything in the output is okay. So - // just store the errors and trigger a stich so that they become visible to - // the outside. - if (!result.ok) { - await this.processingDatabase.transaction(async tx => { - await this.processingDatabase.updateProcessedEntityErrors(tx, { - id, - errors: errorsString, + await Promise.all( + items.map(async item => { + const { id, state, unprocessedEntity } = item; + const result = await this.orchestrator.process({ + entity: unprocessedEntity, + state, }); - }); - await this.stitcher.stitch( - new Set([stringifyEntityRef(unprocessedEntity)]), - ); - return; - } - result.completedEntity.metadata.uid = id; - await this.processingDatabase.transaction(async tx => { - await this.processingDatabase.updateProcessedEntity(tx, { - id, - processedEntity: result.completedEntity, - state: result.state, - errors: errorsString, - relations: result.relations, - deferredEntities: result.deferredEntities, - }); - }); + for (const error of result.errors) { + this.logger.warn(error.message); + } + const errorsString = JSON.stringify( + result.errors.map(e => serializeError(e)), + ); - const setOfThingsToStitch = new Set([ - stringifyEntityRef(result.completedEntity), - ...result.relations.map(relation => stringifyEntityRef(relation.source)), - ]); - await this.stitcher.stitch(setOfThingsToStitch); + // If the result was marked as not OK, it signals that some part of the + // processing pipeline threw an exception. This can happen both as part of + // non-catastrophic things such as due to validation errors, as well as if + // something fatal happens inside the processing for other reasons. In any + // case, this means we can't trust that anything in the output is okay. So + // just store the errors and trigger a stich so that they become visible to + // the outside. + if (!result.ok) { + await this.processingDatabase.transaction(async tx => { + await this.processingDatabase.updateProcessedEntityErrors(tx, { + id, + errors: errorsString, + }); + }); + await this.stitcher.stitch( + new Set([stringifyEntityRef(unprocessedEntity)]), + ); + return; + } + + result.completedEntity.metadata.uid = id; + await this.processingDatabase.transaction(async tx => { + await this.processingDatabase.updateProcessedEntity(tx, { + id, + processedEntity: result.completedEntity, + state: result.state, + errors: errorsString, + relations: result.relations, + deferredEntities: result.deferredEntities, + }); + }); + + const setOfThingsToStitch = new Set([ + stringifyEntityRef(result.completedEntity), + ...result.relations.map(relation => + stringifyEntityRef(relation.source), + ), + ]); + await this.stitcher.stitch(setOfThingsToStitch); + }), + ); } private async wait() { diff --git a/plugins/catalog-backend/src/next/NextEntitiesCatalog.ts b/plugins/catalog-backend/src/next/NextEntitiesCatalog.ts index 13cbf39c7b..9400bf664f 100644 --- a/plugins/catalog-backend/src/next/NextEntitiesCatalog.ts +++ b/plugins/catalog-backend/src/next/NextEntitiesCatalog.ts @@ -101,9 +101,10 @@ export class NextEntitiesCatalog implements EntitiesCatalog { }); } - // TODO: move final_entities to use entity_Ref + // TODO: move final_entities to use entity_ref entitiesQuery = entitiesQuery .select('final_entities.*') + .whereNotNull('final_entities.final_entity') .orderBy('entity_id', 'asc'); const { limit, offset } = parsePagination(request?.pagination); @@ -131,7 +132,7 @@ export class NextEntitiesCatalog implements EntitiesCatalog { } return { - entities: rows.map(e => JSON.parse(e.final_entity)), + entities: rows.map(e => JSON.parse(e.final_entity!)), pageInfo, }; } diff --git a/plugins/catalog-backend/src/next/Stitcher.test.ts b/plugins/catalog-backend/src/next/Stitcher.test.ts index f38af384fa..3f489e82c1 100644 --- a/plugins/catalog-backend/src/next/Stitcher.test.ts +++ b/plugins/catalog-backend/src/next/Stitcher.test.ts @@ -15,6 +15,7 @@ */ import { getVoidLogger } from '@backstage/backend-common'; +import { Entity } from '@backstage/catalog-model'; import { Knex } from 'knex'; import { DatabaseManager } from './database/DatabaseManager'; import { @@ -36,171 +37,170 @@ describe('Stitcher', () => { it('runs the happy path', async () => { const stitcher = new Stitcher(db, logger); + let entities: DbFinalEntitiesRow[]; + let entity: Entity; - await db.transaction(async tx => { - await tx('refresh_state').insert([ - { - entity_id: 'my-id', - entity_ref: 'k:ns/n', - unprocessed_entity: JSON.stringify({}), - processed_entity: JSON.stringify({ - apiVersion: 'a', - kind: 'k', - metadata: { - name: 'n', - namespace: 'ns', - }, - spec: { - k: 'v', - }, - }), - errors: '[]', - next_update_at: tx.fn.now(), - last_discovery_at: tx.fn.now(), - }, - ]); - await tx('refresh_state_references').insert([ - { source_key: 'a', target_entity_ref: 'k:ns/n' }, - ]); - await tx('relations').insert([ - { - originating_entity_id: 'my-id', - source_entity_ref: 'k:ns/n', - type: 'looksAt', - target_entity_ref: 'k:ns/other', - }, - ]); - }); + await db('refresh_state').insert([ + { + entity_id: 'my-id', + entity_ref: 'k:ns/n', + unprocessed_entity: JSON.stringify({}), + processed_entity: JSON.stringify({ + apiVersion: 'a', + kind: 'k', + metadata: { + name: 'n', + namespace: 'ns', + }, + spec: { + k: 'v', + }, + }), + errors: '[]', + next_update_at: db.fn.now(), + last_discovery_at: db.fn.now(), + }, + ]); + await db('refresh_state_references').insert([ + { source_key: 'a', target_entity_ref: 'k:ns/n' }, + ]); + await db('relations').insert([ + { + originating_entity_id: 'my-id', + source_entity_ref: 'k:ns/n', + type: 'looksAt', + target_entity_ref: 'k:ns/other', + }, + ]); + + await db('final_entities') + .insert({ + entity_id: 'my-id', + hash: '', + stitch_ticket: '', + }) + .onConflict('entity_id') + .ignore(); await stitcher.stitch(new Set(['k:ns/n'])); - let firstHash: string; - await db.transaction(async tx => { - const entities = await tx('final_entities'); + entities = await db('final_entities'); - expect(entities.length).toBe(1); - const entity = JSON.parse(entities[0].final_entity); - expect(entity).toEqual({ - relations: [ - { - type: 'looksAt', - target: { - kind: 'k', - namespace: 'ns', - name: 'other', - }, + expect(entities.length).toBe(1); + entity = JSON.parse(entities[0].final_entity!); + expect(entity).toEqual({ + relations: [ + { + type: 'looksAt', + target: { + kind: 'k', + namespace: 'ns', + name: 'other', }, - ], - apiVersion: 'a', - kind: 'k', - metadata: { - name: 'n', - namespace: 'ns', - etag: expect.any(String), - generation: 1, - uid: 'my-id', }, - spec: { - k: 'v', - }, - }); - - expect(entity.metadata.etag).toEqual(entities[0].hash); - firstHash = entities[0].hash; - - const search = await tx('search'); - expect(search).toEqual( - expect.arrayContaining([ - { entity_id: 'my-id', key: 'relations.looksat', value: 'k:ns/other' }, - { entity_id: 'my-id', key: 'apiversion', value: 'a' }, - { entity_id: 'my-id', key: 'kind', value: 'k' }, - { entity_id: 'my-id', key: 'metadata.name', value: 'n' }, - { entity_id: 'my-id', key: 'metadata.namespace', value: 'ns' }, - { entity_id: 'my-id', key: 'metadata.uid', value: 'my-id' }, - { entity_id: 'my-id', key: 'spec.k', value: 'v' }, - ]), - ); + ], + apiVersion: 'a', + kind: 'k', + metadata: { + name: 'n', + namespace: 'ns', + etag: expect.any(String), + generation: 1, + uid: 'my-id', + }, + spec: { + k: 'v', + }, }); + expect(entity.metadata.etag).toEqual(entities[0].hash); + const firstHash = entities[0].hash; + + const search = await db('search'); + expect(search).toEqual( + expect.arrayContaining([ + { entity_id: 'my-id', key: 'relations.looksat', value: 'k:ns/other' }, + { entity_id: 'my-id', key: 'apiversion', value: 'a' }, + { entity_id: 'my-id', key: 'kind', value: 'k' }, + { entity_id: 'my-id', key: 'metadata.name', value: 'n' }, + { entity_id: 'my-id', key: 'metadata.namespace', value: 'ns' }, + { entity_id: 'my-id', key: 'metadata.uid', value: 'my-id' }, + { entity_id: 'my-id', key: 'spec.k', value: 'v' }, + ]), + ); + // Re-stitch without any changes await stitcher.stitch(new Set(['k:ns/n'])); - await db.transaction(async tx => { - const entities = await tx('final_entities'); - expect(entities.length).toBe(1); - const entity = JSON.parse(entities[0].final_entity); - expect(entities[0].hash).toEqual(firstHash); - expect(entity.metadata.etag).toEqual(firstHash); - }); + entities = await db('final_entities'); + expect(entities.length).toBe(1); + entity = JSON.parse(entities[0].final_entity!); + expect(entities[0].hash).toEqual(firstHash); + expect(entity.metadata.etag).toEqual(firstHash); // Now add one more relation and re-stitch - await db.transaction(async tx => { - await tx('relations').insert([ - { - originating_entity_id: 'my-id', - source_entity_ref: 'k:ns/n', - type: 'looksAt', - target_entity_ref: 'k:ns/third', - }, - ]); - }); + await db('relations').insert([ + { + originating_entity_id: 'my-id', + source_entity_ref: 'k:ns/n', + type: 'looksAt', + target_entity_ref: 'k:ns/third', + }, + ]); await stitcher.stitch(new Set(['k:ns/n'])); - await db.transaction(async tx => { - const entities = await tx('final_entities'); + entities = await db('final_entities'); - expect(entities.length).toBe(1); - const entity = JSON.parse(entities[0].final_entity); - expect(entity).toEqual({ - relations: expect.arrayContaining([ - { - type: 'looksAt', - target: { - kind: 'k', - namespace: 'ns', - name: 'other', - }, + expect(entities.length).toBe(1); + entity = JSON.parse(entities[0].final_entity!); + expect(entity).toEqual({ + relations: expect.arrayContaining([ + { + type: 'looksAt', + target: { + kind: 'k', + namespace: 'ns', + name: 'other', }, - { - type: 'looksAt', - target: { - kind: 'k', - namespace: 'ns', - name: 'third', - }, + }, + { + type: 'looksAt', + target: { + kind: 'k', + namespace: 'ns', + name: 'third', }, - ]), - apiVersion: 'a', - kind: 'k', - metadata: { - name: 'n', - namespace: 'ns', - etag: expect.any(String), - generation: 1, - uid: 'my-id', }, - spec: { - k: 'v', - }, - }); - - expect(entities[0].hash).not.toEqual(firstHash); - expect(entities[0].hash).toEqual(entity.metadata.etag); - - const search = await tx('search'); - expect(search).toEqual( - expect.arrayContaining([ - { entity_id: 'my-id', key: 'relations.looksat', value: 'k:ns/other' }, - { entity_id: 'my-id', key: 'relations.looksat', value: 'k:ns/third' }, - { entity_id: 'my-id', key: 'apiversion', value: 'a' }, - { entity_id: 'my-id', key: 'kind', value: 'k' }, - { entity_id: 'my-id', key: 'metadata.name', value: 'n' }, - { entity_id: 'my-id', key: 'metadata.namespace', value: 'ns' }, - { entity_id: 'my-id', key: 'metadata.uid', value: 'my-id' }, - { entity_id: 'my-id', key: 'spec.k', value: 'v' }, - ]), - ); + ]), + apiVersion: 'a', + kind: 'k', + metadata: { + name: 'n', + namespace: 'ns', + etag: expect.any(String), + generation: 1, + uid: 'my-id', + }, + spec: { + k: 'v', + }, }); + + expect(entities[0].hash).not.toEqual(firstHash); + expect(entities[0].hash).toEqual(entity.metadata.etag); + + expect(await db('search')).toEqual( + expect.arrayContaining([ + { entity_id: 'my-id', key: 'relations.looksat', value: 'k:ns/other' }, + { entity_id: 'my-id', key: 'relations.looksat', value: 'k:ns/third' }, + { entity_id: 'my-id', key: 'apiversion', value: 'a' }, + { entity_id: 'my-id', key: 'kind', value: 'k' }, + { entity_id: 'my-id', key: 'metadata.name', value: 'n' }, + { entity_id: 'my-id', key: 'metadata.namespace', value: 'ns' }, + { entity_id: 'my-id', key: 'metadata.uid', value: 'my-id' }, + { entity_id: 'my-id', key: 'spec.k', value: 'v' }, + ]), + ); }); }); diff --git a/plugins/catalog-backend/src/next/Stitcher.ts b/plugins/catalog-backend/src/next/Stitcher.ts index 0281591fb1..8f9514d441 100644 --- a/plugins/catalog-backend/src/next/Stitcher.ts +++ b/plugins/catalog-backend/src/next/Stitcher.ts @@ -20,13 +20,14 @@ import { parseEntityRef, UNSTABLE_EntityStatusItem, } from '@backstage/catalog-model'; -import { ConflictError, SerializedError } from '@backstage/errors'; +import { SerializedError } from '@backstage/errors'; import { createHash } from 'crypto'; import stableStringify from 'fast-json-stable-stringify'; import { Knex } from 'knex'; import { Logger } from 'winston'; -import { Transaction } from '../database'; import { buildEntitySearch, DbSearchRow } from './search'; +import { createTimer } from './util'; +import { v4 as uuid } from 'uuid'; // 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 @@ -37,7 +38,8 @@ const BATCH_SIZE = 50; export type DbFinalEntitiesRow = { entity_id: string; hash: string; - final_entity: string; + stitch_ticket: string; + final_entity?: string; }; function generateStableHash(entity: Entity) { @@ -59,9 +61,23 @@ export class Stitcher { async stitch(entityRefs: Set) { for (const entityRef of entityRefs) { - await this.transaction(async txOpaque => { - const tx = txOpaque as Knex.Transaction; - + const endTimer = createTimer('stitch'); + try { + const ticket = uuid(); + const ticketRows = await this.database( + 'final_entities', + ) + .update('stitch_ticket', ticket) + .whereIn('entity_id', qb => + qb + .select('entity_id') + .from('refresh_state') + .where({ entity_ref: entityRef }), + ); + if (ticketRows === 0) { + // No ticket written. + continue; + } // Selecting from refresh_state and final_entities should yield exactly // one row (except in abnormal cases where the stitch was invoked for // something that didn't exist at all, in which case it's zero rows). @@ -77,7 +93,7 @@ export class Stitcher { previousHash?: string; relationType?: string; relationTarget?: string; - }> = await tx + }> = await this.database .with('incoming_references', function incomingReferences(builder) { return builder .from('refresh_state_references') @@ -95,7 +111,7 @@ export class Stitcher { }) .from('refresh_state') .where({ 'refresh_state.entity_ref': entityRef }) - .crossJoin(tx.raw('incoming_references')) + .crossJoin(this.database.raw('incoming_references')) .leftOuterJoin('final_entities', { 'final_entities.entity_id': 'refresh_state.entity_id', }) @@ -110,10 +126,10 @@ export class Stitcher { // if we emit a relation to something that hasn't been ingested yet. // It's safe to ignore this stitch attempt in that case. if (!result.length) { - this.logger.debug( + this.logger.error( `Unable to stitch ${entityRef}, item does not exist in refresh state table`, ); - return; + continue; } const { @@ -132,7 +148,7 @@ export class Stitcher { this.logger.debug( `Unable to stitch ${entityRef}, the entity has not yet been processed`, ); - return; + continue; } // Grab the processed entity and stitch all of the relevant data into @@ -179,7 +195,7 @@ export class Stitcher { const hash = generateStableHash(entity); if (hash === previousHash) { this.logger.debug(`Skipped stitching of ${entityRef}, no changes`); - return; + continue; } entity.metadata.uid = entityId; @@ -190,56 +206,41 @@ export class Stitcher { entity.metadata.etag = hash; } - await tx('final_entities') - .insert({ - entity_id: entityId, + const rowsChanged = await this.database( + 'final_entities', + ) + .update({ final_entity: JSON.stringify(entity), hash, }) + .where('entity_id', entityId) + .where('stitch_ticket', ticket) .onConflict('entity_id') .merge(['final_entity', 'hash']); + if (rowsChanged.length === 0) { + this.logger.debug( + `Entity ${entityRef} is already processed, skipping write.`, + ); + continue; + } + + // TODO(freben): Search will probably need a similar safeguard against + // race conditions like the final_entities ticket handling above. + // Otherwise, it can be the case that: + // A writes the entity -> + // B writes the entity -> + // B writes search -> + // A writes search const searchEntries = buildEntitySearch(entityId, entity); - await tx('search').where({ entity_id: entityId }).delete(); - await tx.batchInsert('search', searchEntries, BATCH_SIZE); - }); - } - } - - private async transaction( - fn: (tx: Transaction) => Promise, - ): Promise { - try { - let result: T | undefined = undefined; - - await this.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, - isolationLevel: - this.database.client.config.client === 'sqlite3' - ? undefined // sqlite3 only supports serializable transactions, ignoring the isolation level param - : 'serializable', - }, - ); - - return result!; - } catch (e) { - this.logger.debug(`Error during transaction, ${e}`); - - if ( - /SQLITE_CONSTRAINT: UNIQUE/.test(e.message) || - /unique constraint/.test(e.message) - ) { - throw new ConflictError(`Rejected due to a conflicting entity`, e); + await this.database('search') + .where({ entity_id: entityId }) + .delete(); + await this.database.batchInsert('search', searchEntries, BATCH_SIZE); + endTimer(); + } catch (error) { + this.logger.error(`Failed to stitch ${entityRef}, ${error}`); } - - throw e; } } } diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts index 7a0094dbf0..48d6750a84 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts @@ -27,6 +27,7 @@ import { Entity, stringifyEntityRef } from '@backstage/catalog-model'; import * as uuid from 'uuid'; import { getVoidLogger } from '@backstage/backend-common'; import { JsonObject } from '@backstage/config'; +import { DbFinalEntitiesRow } from '../Stitcher'; describe('Default Processing Database', () => { let db: Knex; @@ -115,6 +116,10 @@ describe('Default Processing Database', () => { ); expect(entities[0].cache).toEqual(JSON.stringify(state)); expect(entities[0].errors).toEqual("['something broke']"); + const finalEntities = await db('final_entities') + .where('entity_id', id) + .select(); + expect(finalEntities.length).toBe(1); }); it('removes old relations and stores the new relationships', async () => { diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index cd3ddf037b..b30894c50f 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -32,6 +32,7 @@ import type { Logger } from 'winston'; import { v4 as uuid } from 'uuid'; import { JsonObject } from '@backstage/config'; +import { DbFinalEntitiesRow } from '../Stitcher'; export type DbRefreshStateRow = { entity_id: string; @@ -121,6 +122,15 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { this.deduplicateRelations(relationRows), BATCH_SIZE, ); + + await tx('final_entities') + .insert({ + entity_id: id, + hash: '', + stitch_ticket: '', + }) + .onConflict('entity_id') + .ignore(); } async updateProcessedEntityErrors( From 702b0fc110f44c4df6f35325b13c379b9c98e190 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Wed, 26 May 2021 16:54:28 +0200 Subject: [PATCH 2/4] Remove unused import Signed-off-by: Johan Haals --- plugins/catalog-backend/src/next/Stitcher.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/plugins/catalog-backend/src/next/Stitcher.ts b/plugins/catalog-backend/src/next/Stitcher.ts index 8f9514d441..6958c7c3cf 100644 --- a/plugins/catalog-backend/src/next/Stitcher.ts +++ b/plugins/catalog-backend/src/next/Stitcher.ts @@ -26,7 +26,6 @@ import stableStringify from 'fast-json-stable-stringify'; import { Knex } from 'knex'; import { Logger } from 'winston'; import { buildEntitySearch, DbSearchRow } from './search'; -import { createTimer } from './util'; import { v4 as uuid } from 'uuid'; // The number of items that are sent per batch to the database layer, when From f20c0d8a4ca03af50aa175f3887c296b2b3aa0a5 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Wed, 26 May 2021 16:55:32 +0200 Subject: [PATCH 3/4] chore: remove test code Signed-off-by: Johan Haals --- plugins/catalog-backend/src/next/Stitcher.ts | 2 -- 1 file changed, 2 deletions(-) diff --git a/plugins/catalog-backend/src/next/Stitcher.ts b/plugins/catalog-backend/src/next/Stitcher.ts index 6958c7c3cf..79cbdd3a8e 100644 --- a/plugins/catalog-backend/src/next/Stitcher.ts +++ b/plugins/catalog-backend/src/next/Stitcher.ts @@ -60,7 +60,6 @@ export class Stitcher { async stitch(entityRefs: Set) { for (const entityRef of entityRefs) { - const endTimer = createTimer('stitch'); try { const ticket = uuid(); const ticketRows = await this.database( @@ -236,7 +235,6 @@ export class Stitcher { .where({ entity_id: entityId }) .delete(); await this.database.batchInsert('search', searchEntries, BATCH_SIZE); - endTimer(); } catch (error) { this.logger.error(`Failed to stitch ${entityRef}, ${error}`); } From f33321fd6dba142b7d922a8018fbf46ea82220f0 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Thu, 27 May 2021 13:36:23 +0200 Subject: [PATCH 4/4] Move pre-insertion to final_entity to stitcher Signed-off-by: Johan Haals --- .../next/DefaultCatalogProcessingEngine.ts | 1 + .../catalog-backend/src/next/Stitcher.test.ts | 9 ------ plugins/catalog-backend/src/next/Stitcher.ts | 32 ++++++++++++------- .../DefaultProcessingDatabase.test.ts | 5 --- .../database/DefaultProcessingDatabase.ts | 10 ------ 5 files changed, 21 insertions(+), 36 deletions(-) diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index e908eb95bc..4dda881963 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -110,6 +110,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { return; } + // TODO: replace Promise.all with something more sophisticated for parallel processing. await Promise.all( items.map(async item => { const { id, state, unprocessedEntity } = item; diff --git a/plugins/catalog-backend/src/next/Stitcher.test.ts b/plugins/catalog-backend/src/next/Stitcher.test.ts index 3f489e82c1..a6af6f01a1 100644 --- a/plugins/catalog-backend/src/next/Stitcher.test.ts +++ b/plugins/catalog-backend/src/next/Stitcher.test.ts @@ -73,15 +73,6 @@ describe('Stitcher', () => { }, ]); - await db('final_entities') - .insert({ - entity_id: 'my-id', - hash: '', - stitch_ticket: '', - }) - .onConflict('entity_id') - .ignore(); - await stitcher.stitch(new Set(['k:ns/n'])); entities = await db('final_entities'); diff --git a/plugins/catalog-backend/src/next/Stitcher.ts b/plugins/catalog-backend/src/next/Stitcher.ts index 79cbdd3a8e..3090c4a971 100644 --- a/plugins/catalog-backend/src/next/Stitcher.ts +++ b/plugins/catalog-backend/src/next/Stitcher.ts @@ -27,6 +27,7 @@ import { Knex } from 'knex'; import { Logger } from 'winston'; import { buildEntitySearch, DbSearchRow } from './search'; import { v4 as uuid } from 'uuid'; +import { DbRefreshStateRow } from './database/DefaultProcessingDatabase'; // 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 @@ -61,21 +62,28 @@ export class Stitcher { async stitch(entityRefs: Set) { for (const entityRef of entityRefs) { try { - const ticket = uuid(); - const ticketRows = await this.database( - 'final_entities', + const entityResult = await this.database( + 'refresh_state', ) - .update('stitch_ticket', ticket) - .whereIn('entity_id', qb => - qb - .select('entity_id') - .from('refresh_state') - .where({ entity_ref: entityRef }), - ); - if (ticketRows === 0) { - // No ticket written. + .where({ entity_ref: entityRef }) + .limit(1) + .select('entity_id'); + if (!entityResult.length) { + // Entity does no exist in refresh state table, no stitching required. continue; } + + // Insert stitching ticket that will be compared before inserting the final entity. + const ticket = uuid(); + await this.database('final_entities') + .insert({ + entity_id: entityResult[0].entity_id, + hash: '', + stitch_ticket: ticket, + }) + .onConflict('entity_id') + .merge(['stitch_ticket']); + // Selecting from refresh_state and final_entities should yield exactly // one row (except in abnormal cases where the stitch was invoked for // something that didn't exist at all, in which case it's zero rows). diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts index 48d6750a84..7a0094dbf0 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts @@ -27,7 +27,6 @@ import { Entity, stringifyEntityRef } from '@backstage/catalog-model'; import * as uuid from 'uuid'; import { getVoidLogger } from '@backstage/backend-common'; import { JsonObject } from '@backstage/config'; -import { DbFinalEntitiesRow } from '../Stitcher'; describe('Default Processing Database', () => { let db: Knex; @@ -116,10 +115,6 @@ describe('Default Processing Database', () => { ); expect(entities[0].cache).toEqual(JSON.stringify(state)); expect(entities[0].errors).toEqual("['something broke']"); - const finalEntities = await db('final_entities') - .where('entity_id', id) - .select(); - expect(finalEntities.length).toBe(1); }); it('removes old relations and stores the new relationships', async () => { diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index b30894c50f..cd3ddf037b 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -32,7 +32,6 @@ import type { Logger } from 'winston'; import { v4 as uuid } from 'uuid'; import { JsonObject } from '@backstage/config'; -import { DbFinalEntitiesRow } from '../Stitcher'; export type DbRefreshStateRow = { entity_id: string; @@ -122,15 +121,6 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { this.deduplicateRelations(relationRows), BATCH_SIZE, ); - - await tx('final_entities') - .insert({ - entity_id: id, - hash: '', - stitch_ticket: '', - }) - .onConflict('entity_id') - .ignore(); } async updateProcessedEntityErrors(