Merge pull request #7713 from backstage/freben/stitch-one
Minor rearrangement of `Stitcher` to clarify the scope of one stitch round
This commit is contained in:
Fredrik Adelöw
@@ -0,0 +1,5 @@
|
||||
---
|
||||
'@backstage/plugin-catalog-backend': patch
|
||||
---
|
||||
|
||||
Minor rearrangement of `Stitcher` to clarify the scope of one stitch round
|
||||
@@ -47,194 +47,196 @@ export class Stitcher {
|
||||
async stitch(entityRefs: Set<string>) {
|
||||
for (const entityRef of entityRefs) {
|
||||
try {
|
||||
const entityResult = await this.database<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.
|
||||
continue;
|
||||
}
|
||||
|
||||
// Insert stitching ticket that will be compared before inserting the final entity.
|
||||
const ticket = uuid();
|
||||
await this.database<DbFinalEntitiesRow>('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<DbFinalEntitiesRow>(
|
||||
'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<DbSearchRow>('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<void> {
|
||||
const entityResult = await this.database<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;
|
||||
}
|
||||
|
||||
// Insert stitching ticket that will be compared before inserting the final entity.
|
||||
const ticket = uuid();
|
||||
await this.database<DbFinalEntitiesRow>('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<DbFinalEntitiesRow>(
|
||||
'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<DbSearchRow>('search')
|
||||
.where({ entity_id: entityId })
|
||||
.delete();
|
||||
await this.database.batchInsert('search', searchEntries, BATCH_SIZE);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user