paulononato/backstage
1
0
Fork 0
Code Issues Pull Requests Actions 22 Packages Projects Releases Wiki Activity

Merge pull request #27799 from backstage/freben/deq

Browse Source 
Be more aggressive in dequeueing entities for stitching
This commit is contained in:
Fredrik Adelöw
2024-11-26 10:31:13 +01:00
committed by GitHub
parent 814599ca5a 0b2ccfab57
commit b24221dbf3
3 changed files with 182 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/neat-insects-cough.md
+5
View File
@@ -0,0 +1,5 @@
---
'@backstage/plugin-catalog-backend': patch
---
Be more aggressive in dequeueing entities for stitching
.github/vale/config/vocabularies/Backstage/accept.txt
+2
View File
@@ -101,6 +101,8 @@ Deutsche
dev
devops
devs
dequeue
dequeueing
dialogs
discoverability
Discoverability
plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts
+175 -173
View File
@@ -55,193 +55,195 @@ export async function performStitching(options: {
const { knex, logger, entityRef } = options;
const stitchTicket = options.stitchTicket ?? uuid();
const entityResult = await knex<DbRefreshStateRow>('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<DbFinalEntitiesRow>('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<DbRefreshStateRow>('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<DbFinalEntitiesRow>('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<EntityRelation>(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<EntityRelation>(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<DbFinalEntitiesRow>('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,
// 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<DbFinalEntitiesRow>('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';
}
await knex.transaction(async trx => {
await trx<DbSearchRow>('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<DbSearchRow>('search').where({ entity_id: entityId }).delete();
await trx.batchInsert('search', searchEntries, BATCH_SIZE);
});
await markDeferred();
return 'changed';
}