diff --git a/.changeset/neat-insects-cough.md b/.changeset/neat-insects-cough.md new file mode 100644 index 0000000000..fbdf214edb --- /dev/null +++ b/.changeset/neat-insects-cough.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend': patch +--- + +Be more aggressive in dequeueing entities for stitching diff --git a/.github/vale/config/vocabularies/Backstage/accept.txt b/.github/vale/config/vocabularies/Backstage/accept.txt index a015b97a32..4f5d3e338d 100644 --- a/.github/vale/config/vocabularies/Backstage/accept.txt +++ b/.github/vale/config/vocabularies/Backstage/accept.txt @@ -101,6 +101,8 @@ Deutsche dev devops devs +dequeue +dequeueing dialogs discoverability Discoverability diff --git a/plugins/catalog-backend/src/database/operations/stitcher/buildEntitySearch.test.ts b/plugins/catalog-backend/src/database/operations/stitcher/buildEntitySearch.test.ts index 4241599351..ef21648619 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/buildEntitySearch.test.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/buildEntitySearch.test.ts @@ -236,57 +236,5 @@ describe('buildEntitySearch', () => { }, ]); }); - - it('rejects duplicate keys', () => { - expect(() => - buildEntitySearch('eid', { - relations: [], - apiVersion: 'a', - kind: 'b', - metadata: { - name: 'n', - Namespace: 'ns', - }, - }), - ).toThrow( - `Entity has duplicate keys that vary only in casing, 'metadata.namespace'`, - ); - - expect(() => - buildEntitySearch('eid', { - relations: [ - { - type: 'dup', - targetRef: 'k:ns/a', - }, - { - type: 'DUP', - targetRef: 'k:ns/b', - }, - ], - apiVersion: 'a', - kind: 'b', - metadata: { name: 'n' }, - }), - ).toThrow( - `Entity has duplicate keys that vary only in casing, 'relations.dup'`, - ); - - expect(() => - buildEntitySearch('eid', { - apiVersion: 'a', - kind: 'b', - metadata: { name: 'n' }, - spec: { - owner: 'o', - OWNER: 'o', - lifecycle: 'production', - lifeCycle: 'production', - }, - }), - ).toThrow( - `Entity has duplicate keys that vary only in casing, 'spec.owner', 'spec.lifecycle'`, - ); - }); }); }); diff --git a/plugins/catalog-backend/src/database/operations/stitcher/buildEntitySearch.ts b/plugins/catalog-backend/src/database/operations/stitcher/buildEntitySearch.ts index 53e8d4c1cd..0ea54e29e9 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/buildEntitySearch.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/buildEntitySearch.ts @@ -15,7 +15,6 @@ */ import { DEFAULT_NAMESPACE, Entity } from '@backstage/catalog-model'; -import { InputError } from '@backstage/errors'; import { DbSearchRow } from '../../tables'; // These are excluded in the generic loop, either because they do not make sense @@ -200,23 +199,5 @@ export function buildEntitySearch( }); } - // This validates that there are no keys that vary only in casing, such - // as `spec.foo` and `spec.Foo`. - const keys = new Set(raw.map(r => r.key)); - const lowerKeys = new Set(raw.map(r => r.key.toLocaleLowerCase('en-US'))); - if (keys.size !== lowerKeys.size) { - const difference = []; - for (const key of keys) { - const lower = key.toLocaleLowerCase('en-US'); - if (!lowerKeys.delete(lower)) { - difference.push(lower); - } - } - const badKeys = `'${difference.join("', '")}'`; - throw new InputError( - `Entity has duplicate keys that vary only in casing, ${badKeys}`, - ); - } - return mapToRows(raw, entityId); } diff --git a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts index 198de7209d..353e849176 100644 --- a/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts +++ b/plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts @@ -55,193 +55,191 @@ export async function performStitching(options: { const { knex, logger, entityRef } = options; const stitchTicket = options.stitchTicket ?? uuid(); - const entityResult = await knex('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. - return 'abandoned'; - } + // In deferred mode, the entity is removed from the stitch queue on ANY + // completion, except when an exception is thrown. In the latter case, the + // entity will be retried at a later time. + let removeFromStitchQueueOnCompletion = options.strategy.mode === 'deferred'; - // Insert stitching ticket that will be compared before inserting the final entity. - await knex('final_entities') - .insert({ - entity_id: entityResult[0].entity_id, - hash: '', - entity_ref: entityRef, - stitch_ticket: stitchTicket, - }) - .onConflict('entity_id') - .merge(['stitch_ticket']); + try { + const entityResult = await knex('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. + return 'abandoned'; + } - // 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). - // The join with the temporary incoming_references still gives one row. - const [processedResult, relationsResult] = await Promise.all([ - knex - .with('incoming_references', function incomingReferences(builder) { - return builder - .from('refresh_state_references') - .where({ target_entity_ref: entityRef }) - .count({ count: '*' }); + // Insert stitching ticket that will be compared before inserting the final entity. + await knex('final_entities') + .insert({ + entity_id: entityResult[0].entity_id, + hash: '', + entity_ref: entityRef, + stitch_ticket: stitchTicket, }) - .select({ - entityId: 'refresh_state.entity_id', - processedEntity: 'refresh_state.processed_entity', - errors: 'refresh_state.errors', - incomingReferenceCount: 'incoming_references.count', - previousHash: 'final_entities.hash', - }) - .from('refresh_state') - .where({ 'refresh_state.entity_ref': entityRef }) - .crossJoin(knex.raw('incoming_references')) - .leftOuterJoin('final_entities', { - 'final_entities.entity_id': 'refresh_state.entity_id', - }), - knex - .distinct({ - relationType: 'type', - relationTarget: 'target_entity_ref', - }) - .from('relations') - .where({ source_entity_ref: entityRef }) - .orderBy('relationType', 'asc') - .orderBy('relationTarget', 'asc'), - ]); + .onConflict('entity_id') + .merge(['stitch_ticket']); - // If there were no rows returned, it would mean that there was no - // matching row even in the refresh_state. This can happen for example - // 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 (!processedResult.length) { - logger.debug( - `Unable to stitch ${entityRef}, item does not exist in refresh state table`, - ); - return 'abandoned'; - } + // 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). + // The join with the temporary incoming_references still gives one row. + const [processedResult, relationsResult] = await Promise.all([ + knex + .with('incoming_references', function incomingReferences(builder) { + return builder + .from('refresh_state_references') + .where({ target_entity_ref: entityRef }) + .count({ count: '*' }); + }) + .select({ + entityId: 'refresh_state.entity_id', + processedEntity: 'refresh_state.processed_entity', + errors: 'refresh_state.errors', + incomingReferenceCount: 'incoming_references.count', + previousHash: 'final_entities.hash', + }) + .from('refresh_state') + .where({ 'refresh_state.entity_ref': entityRef }) + .crossJoin(knex.raw('incoming_references')) + .leftOuterJoin('final_entities', { + 'final_entities.entity_id': 'refresh_state.entity_id', + }), + knex + .distinct({ + relationType: 'type', + relationTarget: 'target_entity_ref', + }) + .from('relations') + .where({ source_entity_ref: entityRef }) + .orderBy('relationType', 'asc') + .orderBy('relationTarget', 'asc'), + ]); - const { - entityId, - processedEntity, - errors, - incomingReferenceCount, - previousHash, - } = processedResult[0]; + // If there were no rows returned, it would mean that there was no + // matching row even in the refresh_state. This can happen for example + // 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 (!processedResult.length) { + logger.debug( + `Unable to stitch ${entityRef}, item does not exist in refresh state table`, + ); + return 'abandoned'; + } - // If there was no processed entity in place, the target hasn't been - // through the processing steps yet. It's safe to ignore this stitch - // attempt in that case, since another stitch will be triggered when - // that processing has finished. - if (!processedEntity) { - logger.debug( - `Unable to stitch ${entityRef}, the entity has not yet been processed`, - ); - return 'abandoned'; - } + const { + entityId, + processedEntity, + errors, + incomingReferenceCount, + previousHash, + } = processedResult[0]; - // Grab the processed entity and stitch all of the relevant data into - // it - const entity = JSON.parse(processedEntity) as AlphaEntity; - const isOrphan = Number(incomingReferenceCount) === 0; - let statusItems: EntityStatusItem[] = []; + // If there was no processed entity in place, the target hasn't been + // through the processing steps yet. It's safe to ignore this stitch + // attempt in that case, since another stitch will be triggered when + // that processing has finished. + if (!processedEntity) { + logger.debug( + `Unable to stitch ${entityRef}, the entity has not yet been processed`, + ); + return 'abandoned'; + } - if (isOrphan) { - logger.debug(`${entityRef} is an orphan`); - entity.metadata.annotations = { - ...entity.metadata.annotations, - ['backstage.io/orphan']: 'true', - }; - } - if (errors) { - const parsedErrors = JSON.parse(errors) as SerializedError[]; - if (Array.isArray(parsedErrors) && parsedErrors.length) { - statusItems = parsedErrors.map(e => ({ - type: ENTITY_STATUS_CATALOG_PROCESSING_TYPE, - level: 'error', - message: `${e.name}: ${e.message}`, - error: e, + // Grab the processed entity and stitch all of the relevant data into + // it + const entity = JSON.parse(processedEntity) as AlphaEntity; + const isOrphan = Number(incomingReferenceCount) === 0; + let statusItems: EntityStatusItem[] = []; + + if (isOrphan) { + logger.debug(`${entityRef} is an orphan`); + entity.metadata.annotations = { + ...entity.metadata.annotations, + ['backstage.io/orphan']: 'true', + }; + } + if (errors) { + const parsedErrors = JSON.parse(errors) as SerializedError[]; + if (Array.isArray(parsedErrors) && parsedErrors.length) { + statusItems = parsedErrors.map(e => ({ + type: ENTITY_STATUS_CATALOG_PROCESSING_TYPE, + level: 'error', + message: `${e.name}: ${e.message}`, + error: e, + })); + } + } + // We opt to do this check here as we otherwise can't guarantee that it will be run after all processors + for (const annotation of [ANNOTATION_VIEW_URL, ANNOTATION_EDIT_URL]) { + const value = entity.metadata.annotations?.[annotation]; + if (typeof value === 'string' && scriptProtocolPattern.test(value)) { + entity.metadata.annotations![annotation] = + 'https://backstage.io/annotation-rejected-for-security-reasons'; + } + } + + // TODO: entityRef is lower case and should be uppercase in the final + // result + entity.relations = relationsResult + .filter(row => row.relationType /* exclude null row, if relevant */) + .map(row => ({ + type: row.relationType!, + targetRef: row.relationTarget!, })); + if (statusItems.length) { + entity.status = { + ...entity.status, + items: [...(entity.status?.items ?? []), ...statusItems], + }; } - } - // We opt to do this check here as we otherwise can't guarantee that it will be run after all processors - for (const annotation of [ANNOTATION_VIEW_URL, ANNOTATION_EDIT_URL]) { - const value = entity.metadata.annotations?.[annotation]; - if (typeof value === 'string' && scriptProtocolPattern.test(value)) { - entity.metadata.annotations![annotation] = - 'https://backstage.io/annotation-rejected-for-security-reasons'; + + // If the output entity was actually not changed, just abort + const hash = generateStableHash(entity); + if (hash === previousHash) { + logger.debug(`Skipped stitching of ${entityRef}, no changes`); + return 'unchanged'; } - } - // TODO: entityRef is lower case and should be uppercase in the final - // result - entity.relations = relationsResult - .filter(row => row.relationType /* exclude null row, if relevant */) - .map(row => ({ - type: row.relationType!, - targetRef: row.relationTarget!, - })); - if (statusItems.length) { - entity.status = { - ...entity.status, - items: [...(entity.status?.items ?? []), ...statusItems], - }; - } - - // If the output entity was actually not changed, just abort - const hash = generateStableHash(entity); - if (hash === previousHash) { - logger.debug(`Skipped stitching of ${entityRef}, no changes`); - return 'unchanged'; - } - - entity.metadata.uid = entityId; - if (!entity.metadata.etag) { - // If the original data source did not have its own etag handling, - // use the hash as a good-quality etag - entity.metadata.etag = hash; - } - - // This may throw if the entity is invalid, so we call it before - // the final_entities write, even though we may end up not needing - // to write the search index. - const searchEntries = buildEntitySearch(entityId, entity); - - const amountOfRowsChanged = await knex('final_entities') - .update({ - final_entity: JSON.stringify(entity), - hash, - last_updated_at: knex.fn.now(), - entity_ref: entityRef, - }) - .where('entity_id', entityId) - .where('stitch_ticket', stitchTicket) - .onConflict('entity_id') - .merge(['final_entity', 'hash', 'last_updated_at']); - - const markDeferred = async () => { - if (options.strategy.mode !== 'deferred') { - return; + entity.metadata.uid = entityId; + if (!entity.metadata.etag) { + // If the original data source did not have its own etag handling, + // use the hash as a good-quality etag + entity.metadata.etag = hash; } - await markDeferredStitchCompleted({ - knex: knex, - entityRef, - stitchTicket, + + const amountOfRowsChanged = await knex('final_entities') + .update({ + final_entity: JSON.stringify(entity), + hash, + last_updated_at: knex.fn.now(), + }) + .where('entity_id', entityId) + .where('stitch_ticket', stitchTicket); + + if (amountOfRowsChanged === 0) { + logger.debug(`Entity ${entityRef} is already stitched, skipping write.`); + return 'abandoned'; + } + + const searchEntries = buildEntitySearch(entityId, entity); + await knex.transaction(async trx => { + await trx('search').where({ entity_id: entityId }).delete(); + await trx.batchInsert('search', searchEntries, BATCH_SIZE); }); - }; - if (amountOfRowsChanged === 0) { - logger.debug(`Entity ${entityRef} is already stitched, skipping write.`); - await markDeferred(); - return 'abandoned'; + return 'changed'; + } catch (error) { + removeFromStitchQueueOnCompletion = false; + throw error; + } finally { + if (removeFromStitchQueueOnCompletion) { + await markDeferredStitchCompleted({ + knex: knex, + entityRef, + stitchTicket, + }); + } } - - await knex.transaction(async trx => { - await trx('search').where({ entity_id: entityId }).delete(); - await trx.batchInsert('search', searchEntries, BATCH_SIZE); - }); - - await markDeferred(); - return 'changed'; }