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

Merge pull request #5820 from backstage/mob/stitch-ticket

Browse Source 
catalog/next: Replace stitch transactions with ticket
This commit is contained in:
Johan Haals
2021-05-27 14:12:22 +02:00
committed by GitHub
parent 6665e16c7d f33321fd6d
commit baee1ed980
5 changed files with 261 additions and 250 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/catalog-backend/migrationsv2/20210302150147_refresh_state.js
+10 -4
View File
@@ -92,8 +92,14 @@ exports.up = async function up(knex) {
'Stable hash of the entity data, to be used for caching and avoiding redundant work',
);
table
.text('final_entity')
.text('stitch_ticket')
.notNullable()
.comment(
'A random value representing a unique stitch attempt ticket, that gets updated each time that a stitching attempt is made on the entity',
);
table
.text('final_entity')
.nullable()
.comment('The JSON encoded final entity');
table.index('entity_id', 'final_entities_entity_id_idx');
});
@@ -192,9 +198,9 @@ exports.up = async function up(knex) {
*/
exports.down = async function down(knex) {
await knex.schema.alterTable('refresh_state_references', table => {
table.dropIndex([], 'refresh_state_references_source_special_key_idx');
table.dropIndex([], 'refresh_state_references_source_entity_id_idx');
table.dropIndex([], 'refresh_state_references_target_entity_id_idx');
table.dropIndex([], 'refresh_state_references_source_key_idx');
table.dropIndex([], 'refresh_state_references_source_entity_ref_idx');
table.dropIndex([], 'refresh_state_references_target_entity_ref_idx');
});
await knex.schema.alterTable('refresh_state', table => {
table.dropUnique([], 'refresh_state_entity_ref_uniq');
plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts
+54 -47
View File
@@ -110,56 +110,63 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
return;
}
const { id, state, unprocessedEntity } = items[0];
const result = await this.orchestrator.process({
entity: unprocessedEntity,
state,
});
for (const error of result.errors) {
this.logger.warn(error.message);
}
const errorsString = JSON.stringify(
result.errors.map(e => serializeError(e)),
);
// If the result was marked as not OK, it signals that some part of the
// processing pipeline threw an exception. This can happen both as part of
// non-catastrophic things such as due to validation errors, as well as if
// something fatal happens inside the processing for other reasons. In any
// case, this means we can't trust that anything in the output is okay. So
// just store the errors and trigger a stich so that they become visible to
// the outside.
if (!result.ok) {
await this.processingDatabase.transaction(async tx => {
await this.processingDatabase.updateProcessedEntityErrors(tx, {
id,
errors: errorsString,
// TODO: replace Promise.all with something more sophisticated for parallel processing.
await Promise.all(
items.map(async item => {
const { id, state, unprocessedEntity } = item;
const result = await this.orchestrator.process({
entity: unprocessedEntity,
state,
});
});
await this.stitcher.stitch(
new Set([stringifyEntityRef(unprocessedEntity)]),
);
return;
}
result.completedEntity.metadata.uid = id;
await this.processingDatabase.transaction(async tx => {
await this.processingDatabase.updateProcessedEntity(tx, {
id,
processedEntity: result.completedEntity,
state: result.state,
errors: errorsString,
relations: result.relations,
deferredEntities: result.deferredEntities,
});
});
for (const error of result.errors) {
this.logger.warn(error.message);
}
const errorsString = JSON.stringify(
result.errors.map(e => serializeError(e)),
);
const setOfThingsToStitch = new Set<string>([
stringifyEntityRef(result.completedEntity),
...result.relations.map(relation => stringifyEntityRef(relation.source)),
]);
await this.stitcher.stitch(setOfThingsToStitch);
// If the result was marked as not OK, it signals that some part of the
// processing pipeline threw an exception. This can happen both as part of
// non-catastrophic things such as due to validation errors, as well as if
// something fatal happens inside the processing for other reasons. In any
// case, this means we can't trust that anything in the output is okay. So
// just store the errors and trigger a stich so that they become visible to
// the outside.
if (!result.ok) {
await this.processingDatabase.transaction(async tx => {
await this.processingDatabase.updateProcessedEntityErrors(tx, {
id,
errors: errorsString,
});
});
await this.stitcher.stitch(
new Set([stringifyEntityRef(unprocessedEntity)]),
);
return;
}
result.completedEntity.metadata.uid = id;
await this.processingDatabase.transaction(async tx => {
await this.processingDatabase.updateProcessedEntity(tx, {
id,
processedEntity: result.completedEntity,
state: result.state,
errors: errorsString,
relations: result.relations,
deferredEntities: result.deferredEntities,
});
});
const setOfThingsToStitch = new Set<string>([
stringifyEntityRef(result.completedEntity),
...result.relations.map(relation =>
stringifyEntityRef(relation.source),
),
]);
await this.stitcher.stitch(setOfThingsToStitch);
}),
);
}
private async wait() {
plugins/catalog-backend/src/next/NextEntitiesCatalog.ts
+3 -2
View File
@@ -101,9 +101,10 @@ export class NextEntitiesCatalog implements EntitiesCatalog {
});
}
// TODO: move final_entities to use entity_Ref
// TODO: move final_entities to use entity_ref
entitiesQuery = entitiesQuery
.select('final_entities.*')
.whereNotNull('final_entities.final_entity')
.orderBy('entity_id', 'asc');
const { limit, offset } = parsePagination(request?.pagination);
@@ -131,7 +132,7 @@ export class NextEntitiesCatalog implements EntitiesCatalog {
}
return {
entities: rows.map(e => JSON.parse(e.final_entity)),
entities: rows.map(e => JSON.parse(e.final_entity!)),
pageInfo,
};
}
plugins/catalog-backend/src/next/Stitcher.test.ts
+135 -144
View File
@@ -15,6 +15,7 @@
*/
import { getVoidLogger } from '@backstage/backend-common';
import { Entity } from '@backstage/catalog-model';
import { Knex } from 'knex';
import { DatabaseManager } from './database/DatabaseManager';
import {
@@ -36,171 +37,161 @@ describe('Stitcher', () => {
it('runs the happy path', async () => {
const stitcher = new Stitcher(db, logger);
let entities: DbFinalEntitiesRow[];
let entity: Entity;
await db.transaction(async tx => {
await tx<DbRefreshStateRow>('refresh_state').insert([
{
entity_id: 'my-id',
entity_ref: 'k:ns/n',
unprocessed_entity: JSON.stringify({}),
processed_entity: JSON.stringify({
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
},
spec: {
k: 'v',
},
}),
errors: '[]',
next_update_at: tx.fn.now(),
last_discovery_at: tx.fn.now(),
},
]);
await tx<DbRefreshStateReferencesRow>('refresh_state_references').insert([
{ source_key: 'a', target_entity_ref: 'k:ns/n' },
]);
await tx<DbRelationsRow>('relations').insert([
{
originating_entity_id: 'my-id',
source_entity_ref: 'k:ns/n',
type: 'looksAt',
target_entity_ref: 'k:ns/other',
},
]);
});
await db<DbRefreshStateRow>('refresh_state').insert([
{
entity_id: 'my-id',
entity_ref: 'k:ns/n',
unprocessed_entity: JSON.stringify({}),
processed_entity: JSON.stringify({
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
},
spec: {
k: 'v',
},
}),
errors: '[]',
next_update_at: db.fn.now(),
last_discovery_at: db.fn.now(),
},
]);
await db<DbRefreshStateReferencesRow>('refresh_state_references').insert([
{ source_key: 'a', target_entity_ref: 'k:ns/n' },
]);
await db<DbRelationsRow>('relations').insert([
{
originating_entity_id: 'my-id',
source_entity_ref: 'k:ns/n',
type: 'looksAt',
target_entity_ref: 'k:ns/other',
},
]);
await stitcher.stitch(new Set(['k:ns/n']));
let firstHash: string;
await db.transaction(async tx => {
const entities = await tx<DbFinalEntitiesRow>('final_entities');
entities = await db<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
const entity = JSON.parse(entities[0].final_entity);
expect(entity).toEqual({
relations: [
{
type: 'looksAt',
target: {
kind: 'k',
namespace: 'ns',
name: 'other',
},
expect(entities.length).toBe(1);
entity = JSON.parse(entities[0].final_entity!);
expect(entity).toEqual({
relations: [
{
type: 'looksAt',
target: {
kind: 'k',
namespace: 'ns',
name: 'other',
},
],
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
etag: expect.any(String),
generation: 1,
uid: 'my-id',
},
spec: {
k: 'v',
},
});
expect(entity.metadata.etag).toEqual(entities[0].hash);
firstHash = entities[0].hash;
const search = await tx<DbSearchRow>('search');
expect(search).toEqual(
expect.arrayContaining([
{ entity_id: 'my-id', key: 'relations.looksat', value: 'k:ns/other' },
{ entity_id: 'my-id', key: 'apiversion', value: 'a' },
{ entity_id: 'my-id', key: 'kind', value: 'k' },
{ entity_id: 'my-id', key: 'metadata.name', value: 'n' },
{ entity_id: 'my-id', key: 'metadata.namespace', value: 'ns' },
{ entity_id: 'my-id', key: 'metadata.uid', value: 'my-id' },
{ entity_id: 'my-id', key: 'spec.k', value: 'v' },
]),
);
],
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
etag: expect.any(String),
generation: 1,
uid: 'my-id',
},
spec: {
k: 'v',
},
});
expect(entity.metadata.etag).toEqual(entities[0].hash);
const firstHash = entities[0].hash;
const search = await db<DbSearchRow>('search');
expect(search).toEqual(
expect.arrayContaining([
{ entity_id: 'my-id', key: 'relations.looksat', value: 'k:ns/other' },
{ entity_id: 'my-id', key: 'apiversion', value: 'a' },
{ entity_id: 'my-id', key: 'kind', value: 'k' },
{ entity_id: 'my-id', key: 'metadata.name', value: 'n' },
{ entity_id: 'my-id', key: 'metadata.namespace', value: 'ns' },
{ entity_id: 'my-id', key: 'metadata.uid', value: 'my-id' },
{ entity_id: 'my-id', key: 'spec.k', value: 'v' },
]),
);
// Re-stitch without any changes
await stitcher.stitch(new Set(['k:ns/n']));
await db.transaction(async tx => {
const entities = await tx<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
const entity = JSON.parse(entities[0].final_entity);
expect(entities[0].hash).toEqual(firstHash);
expect(entity.metadata.etag).toEqual(firstHash);
});
entities = await db<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
entity = JSON.parse(entities[0].final_entity!);
expect(entities[0].hash).toEqual(firstHash);
expect(entity.metadata.etag).toEqual(firstHash);
// Now add one more relation and re-stitch
await db.transaction(async tx => {
await tx<DbRelationsRow>('relations').insert([
{
originating_entity_id: 'my-id',
source_entity_ref: 'k:ns/n',
type: 'looksAt',
target_entity_ref: 'k:ns/third',
},
]);
});
await db<DbRelationsRow>('relations').insert([
{
originating_entity_id: 'my-id',
source_entity_ref: 'k:ns/n',
type: 'looksAt',
target_entity_ref: 'k:ns/third',
},
]);
await stitcher.stitch(new Set(['k:ns/n']));
await db.transaction(async tx => {
const entities = await tx<DbFinalEntitiesRow>('final_entities');
entities = await db<DbFinalEntitiesRow>('final_entities');
expect(entities.length).toBe(1);
const entity = JSON.parse(entities[0].final_entity);
expect(entity).toEqual({
relations: expect.arrayContaining([
{
type: 'looksAt',
target: {
kind: 'k',
namespace: 'ns',
name: 'other',
},
expect(entities.length).toBe(1);
entity = JSON.parse(entities[0].final_entity!);
expect(entity).toEqual({
relations: expect.arrayContaining([
{
type: 'looksAt',
target: {
kind: 'k',
namespace: 'ns',
name: 'other',
},
{
type: 'looksAt',
target: {
kind: 'k',
namespace: 'ns',
name: 'third',
},
},
{
type: 'looksAt',
target: {
kind: 'k',
namespace: 'ns',
name: 'third',
},
]),
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
etag: expect.any(String),
generation: 1,
uid: 'my-id',
},
spec: {
k: 'v',
},
});
expect(entities[0].hash).not.toEqual(firstHash);
expect(entities[0].hash).toEqual(entity.metadata.etag);
const search = await tx<DbSearchRow>('search');
expect(search).toEqual(
expect.arrayContaining([
{ entity_id: 'my-id', key: 'relations.looksat', value: 'k:ns/other' },
{ entity_id: 'my-id', key: 'relations.looksat', value: 'k:ns/third' },
{ entity_id: 'my-id', key: 'apiversion', value: 'a' },
{ entity_id: 'my-id', key: 'kind', value: 'k' },
{ entity_id: 'my-id', key: 'metadata.name', value: 'n' },
{ entity_id: 'my-id', key: 'metadata.namespace', value: 'ns' },
{ entity_id: 'my-id', key: 'metadata.uid', value: 'my-id' },
{ entity_id: 'my-id', key: 'spec.k', value: 'v' },
]),
);
]),
apiVersion: 'a',
kind: 'k',
metadata: {
name: 'n',
namespace: 'ns',
etag: expect.any(String),
generation: 1,
uid: 'my-id',
},
spec: {
k: 'v',
},
});
expect(entities[0].hash).not.toEqual(firstHash);
expect(entities[0].hash).toEqual(entity.metadata.etag);
expect(await db<DbSearchRow>('search')).toEqual(
expect.arrayContaining([
{ entity_id: 'my-id', key: 'relations.looksat', value: 'k:ns/other' },
{ entity_id: 'my-id', key: 'relations.looksat', value: 'k:ns/third' },
{ entity_id: 'my-id', key: 'apiversion', value: 'a' },
{ entity_id: 'my-id', key: 'kind', value: 'k' },
{ entity_id: 'my-id', key: 'metadata.name', value: 'n' },
{ entity_id: 'my-id', key: 'metadata.namespace', value: 'ns' },
{ entity_id: 'my-id', key: 'metadata.uid', value: 'my-id' },
{ entity_id: 'my-id', key: 'spec.k', value: 'v' },
]),
);
});
});
plugins/catalog-backend/src/next/Stitcher.ts
+59 -53
View File
@@ -20,13 +20,14 @@ import {
parseEntityRef,
UNSTABLE_EntityStatusItem,
} from '@backstage/catalog-model';
import { ConflictError, SerializedError } from '@backstage/errors';
import { SerializedError } from '@backstage/errors';
import { createHash } from 'crypto';
import stableStringify from 'fast-json-stable-stringify';
import { Knex } from 'knex';
import { Logger } from 'winston';
import { Transaction } from '../database';
import { buildEntitySearch, DbSearchRow } from './search';
import { v4 as uuid } from 'uuid';
import { DbRefreshStateRow } from './database/DefaultProcessingDatabase';
// The number of items that are sent per batch to the database layer, when
// doing .batchInsert calls to knex. This needs to be low enough to not cause
@@ -37,7 +38,8 @@ const BATCH_SIZE = 50;
export type DbFinalEntitiesRow = {
entity_id: string;
hash: string;
final_entity: string;
stitch_ticket: string;
final_entity?: string;
};
function generateStableHash(entity: Entity) {
@@ -59,8 +61,28 @@ export class Stitcher {
async stitch(entityRefs: Set<string>) {
for (const entityRef of entityRefs) {
await this.transaction(async txOpaque => {
const tx = txOpaque as Knex.Transaction;
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
@@ -77,7 +99,7 @@ export class Stitcher {
previousHash?: string;
relationType?: string;
relationTarget?: string;
}> = await tx
}> = await this.database
.with('incoming_references', function incomingReferences(builder) {
return builder
.from('refresh_state_references')
@@ -95,7 +117,7 @@ export class Stitcher {
})
.from('refresh_state')
.where({ 'refresh_state.entity_ref': entityRef })
.crossJoin(tx.raw('incoming_references'))
.crossJoin(this.database.raw('incoming_references'))
.leftOuterJoin('final_entities', {
'final_entities.entity_id': 'refresh_state.entity_id',
})
@@ -110,10 +132,10 @@ export class Stitcher {
// 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.debug(
this.logger.error(
`Unable to stitch ${entityRef}, item does not exist in refresh state table`,
);
return;
continue;
}
const {
@@ -132,7 +154,7 @@ export class Stitcher {
this.logger.debug(
`Unable to stitch ${entityRef}, the entity has not yet been processed`,
);
return;
continue;
}
// Grab the processed entity and stitch all of the relevant data into
@@ -179,7 +201,7 @@ export class Stitcher {
const hash = generateStableHash(entity);
if (hash === previousHash) {
this.logger.debug(`Skipped stitching of ${entityRef}, no changes`);
return;
continue;
}
entity.metadata.uid = entityId;
@@ -190,56 +212,40 @@ export class Stitcher {
entity.metadata.etag = hash;
}
await tx<DbFinalEntitiesRow>('final_entities')
.insert({
entity_id: entityId,
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 tx<DbSearchRow>('search').where({ entity_id: entityId }).delete();
await tx.batchInsert('search', searchEntries, BATCH_SIZE);
});
}
}
private async transaction<T>(
fn: (tx: Transaction) => Promise<T>,
): Promise<T> {
try {
let result: T | undefined = undefined;
await this.database.transaction(
async tx => {
// We can't return here, as knex swallows the return type in case the transaction is rolled back:
// https://github.com/knex/knex/blob/e37aeaa31c8ef9c1b07d2e4d3ec6607e557d800d/lib/transaction.js#L136
result = await fn(tx);
},
{
// If we explicitly trigger a rollback, don't fail.
doNotRejectOnRollback: true,
isolationLevel:
this.database.client.config.client === 'sqlite3'
? undefined // sqlite3 only supports serializable transactions, ignoring the isolation level param
: 'serializable',
},
);
return result!;
} catch (e) {
this.logger.debug(`Error during transaction, ${e}`);
if (
/SQLITE_CONSTRAINT: UNIQUE/.test(e.message) ||
/unique constraint/.test(e.message)
) {
throw new ConflictError(`Rejected due to a conflicting entity`, e);
await this.database<DbSearchRow>('search')
.where({ entity_id: entityId })
.delete();
await this.database.batchInsert('search', searchEntries, BATCH_SIZE);
} catch (error) {
this.logger.error(`Failed to stitch ${entityRef}, ${error}`);
}
throw e;
}
}
}