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..4dda881963 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -110,56 +110,63 @@ 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, + // TODO: replace Promise.all with something more sophisticated for parallel processing. + 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..a6af6f01a1 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,161 @@ 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 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..3090c4a971 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 { 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 @@ -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,8 +61,28 @@ export class Stitcher { async stitch(entityRefs: Set) { for (const entityRef of entityRefs) { - await this.transaction(async txOpaque => { - const tx = txOpaque as Knex.Transaction; + try { + const entityResult = await this.database( + 'refresh_state', + ) + .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 @@ -77,7 +99,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 +117,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 +132,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 +154,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 +201,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 +212,40 @@ 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); + } catch (error) { + this.logger.error(`Failed to stitch ${entityRef}, ${error}`); } - - throw e; } } }