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

Be more aggressive in dequeueing entities for stitching

Browse Source 
Signed-off-by: Fredrik Adelöw <freben@gmail.com>
This commit is contained in:
Fredrik Adelöw
2024-11-24 13:58:14 +01:00
parent 54916f2446
commit 56511ba1ac
5 changed files with 178 additions and 244 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/buildEntitySearch.test.ts
-52
View File
@@ -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'`,
);
});
});
});
plugins/catalog-backend/src/database/operations/stitcher/buildEntitySearch.ts
-19
View File
@@ -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);
}
plugins/catalog-backend/src/database/operations/stitcher/performStitching.ts
+171 -173
View File
@@ -55,193 +55,191 @@ 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,
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';
}
const searchEntries = buildEntitySearch(entityId, entity);
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';
}