From 1f62d1cbe9185b0253427ab5602b2530b87a369b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Wed, 20 Oct 2021 20:30:27 +0200 Subject: [PATCH] Minor rearrangement of `Stitcher` to clarify the scope of one stitch round MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- .changeset/hot-wolves-share.md | 5 + .../catalog-backend/src/stitching/Stitcher.ts | 372 +++++++++--------- 2 files changed, 192 insertions(+), 185 deletions(-) create mode 100644 .changeset/hot-wolves-share.md diff --git a/.changeset/hot-wolves-share.md b/.changeset/hot-wolves-share.md new file mode 100644 index 0000000000..a6294b8c9c --- /dev/null +++ b/.changeset/hot-wolves-share.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend': patch +--- + +Minor rearrangement of `Stitcher` to clarify the scope of one stitch round diff --git a/plugins/catalog-backend/src/stitching/Stitcher.ts b/plugins/catalog-backend/src/stitching/Stitcher.ts index 0aa39f5ac8..1d40340b3e 100644 --- a/plugins/catalog-backend/src/stitching/Stitcher.ts +++ b/plugins/catalog-backend/src/stitching/Stitcher.ts @@ -47,194 +47,196 @@ export class Stitcher { async stitch(entityRefs: Set) { for (const entityRef of entityRefs) { 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 - // 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. - // The only result set "expanding" join is the one with relations, so - // the output should be at least one row (if zero or one relations were - // found), or at most the same number of rows as relations. - const result: Array<{ - entityId: string; - processedEntity?: string; - errors: string; - incomingReferenceCount: string | number; - previousHash?: string; - relationType?: string; - relationTarget?: string; - }> = await this.database - .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', - relationType: 'relations.type', - relationTarget: 'relations.target_entity_ref', - }) - .from('refresh_state') - .where({ 'refresh_state.entity_ref': entityRef }) - .crossJoin(this.database.raw('incoming_references')) - .leftOuterJoin('final_entities', { - 'final_entities.entity_id': 'refresh_state.entity_id', - }) - .leftOuterJoin('relations', { - 'relations.source_entity_ref': 'refresh_state.entity_ref', - }) - .orderBy('relationType', 'asc') - .orderBy('relationTarget', 'asc'); - - // 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 (!result.length) { - this.logger.error( - `Unable to stitch ${entityRef}, item does not exist in refresh state table`, - ); - continue; - } - - const { - entityId, - processedEntity, - errors, - incomingReferenceCount, - previousHash, - } = result[0]; - - // 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) { - this.logger.debug( - `Unable to stitch ${entityRef}, the entity has not yet been processed`, - ); - continue; - } - - // Grab the processed entity and stitch all of the relevant data into - // it - const entity = JSON.parse(processedEntity) as Entity; - const isOrphan = Number(incomingReferenceCount) === 0; - let statusItems: UNSTABLE_EntityStatusItem[] = []; - - if (isOrphan) { - this.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, - })); - } - } - - // TODO: entityRef is lower case and should be uppercase in the final - // result - const uniqueRelationRows = uniqBy( - result, - r => `${r.relationType}:${r.relationTarget}`, - ); - entity.relations = uniqueRelationRows - .filter(row => row.relationType /* exclude null row, if relevant */) - .map(row => ({ - type: row.relationType!, - target: parseEntityRef(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) { - this.logger.debug(`Skipped stitching of ${entityRef}, no changes`); - continue; - } - - entity.metadata.uid = entityId; - entity.metadata.generation = 1; - 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; - } - - 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 this.database('search') - .where({ entity_id: entityId }) - .delete(); - await this.database.batchInsert('search', searchEntries, BATCH_SIZE); + await this.stitchOne(entityRef); } catch (error) { this.logger.error(`Failed to stitch ${entityRef}, ${error}`); } } } + + private async stitchOne(entityRef: string): Promise { + 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. + return; + } + + // 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). + // The join with the temporary incoming_references still gives one row. + // The only result set "expanding" join is the one with relations, so + // the output should be at least one row (if zero or one relations were + // found), or at most the same number of rows as relations. + const result: Array<{ + entityId: string; + processedEntity?: string; + errors: string; + incomingReferenceCount: string | number; + previousHash?: string; + relationType?: string; + relationTarget?: string; + }> = await this.database + .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', + relationType: 'relations.type', + relationTarget: 'relations.target_entity_ref', + }) + .from('refresh_state') + .where({ 'refresh_state.entity_ref': entityRef }) + .crossJoin(this.database.raw('incoming_references')) + .leftOuterJoin('final_entities', { + 'final_entities.entity_id': 'refresh_state.entity_id', + }) + .leftOuterJoin('relations', { + 'relations.source_entity_ref': 'refresh_state.entity_ref', + }) + .orderBy('relationType', 'asc') + .orderBy('relationTarget', 'asc'); + + // 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 (!result.length) { + this.logger.error( + `Unable to stitch ${entityRef}, item does not exist in refresh state table`, + ); + return; + } + + const { + entityId, + processedEntity, + errors, + incomingReferenceCount, + previousHash, + } = result[0]; + + // 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) { + this.logger.debug( + `Unable to stitch ${entityRef}, the entity has not yet been processed`, + ); + return; + } + + // Grab the processed entity and stitch all of the relevant data into + // it + const entity = JSON.parse(processedEntity) as Entity; + const isOrphan = Number(incomingReferenceCount) === 0; + let statusItems: UNSTABLE_EntityStatusItem[] = []; + + if (isOrphan) { + this.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, + })); + } + } + + // TODO: entityRef is lower case and should be uppercase in the final + // result + const uniqueRelationRows = uniqBy( + result, + r => `${r.relationType}:${r.relationTarget}`, + ); + entity.relations = uniqueRelationRows + .filter(row => row.relationType /* exclude null row, if relevant */) + .map(row => ({ + type: row.relationType!, + target: parseEntityRef(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) { + this.logger.debug(`Skipped stitching of ${entityRef}, no changes`); + return; + } + + entity.metadata.uid = entityId; + entity.metadata.generation = 1; + 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; + } + + 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.`, + ); + return; + } + + // 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 this.database('search') + .where({ entity_id: entityId }) + .delete(); + await this.database.batchInsert('search', searchEntries, BATCH_SIZE); + } }