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

Write relations directly as part of batch add / update of entities.

Browse Source 
This implies a slight change of the CommonDatabase contract.

The huge LDAP dataset is now down to 10s to write, both on postgres and sqlite.
This commit is contained in:
Fredrik Adelöw
2020-12-13 15:03:39 +01:00
parent 5153e3eed9
commit 6b37c95bfb
6 changed files with 168 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/purple-cycles-switch.md
+26
View File
@@ -0,0 +1,26 @@
---
'@backstage/plugin-catalog-backend': minor
---
Write relations directly as part of batch add / update of entities.
Slight change of the `CommonDatabase` contract:
## `addEntity` removed
This method was unused by the core, and rendered unnecessary when `addEntities`
exists.
If you were a user of `addEntity`, please call `addEntities` instead, with an
array of one element.
## `DbEntityRequest` has a new field `relations`
This is the structure that is passed to `addEntities` and `updateEntity`. It
used to be the case that you needed to call `setRelations` separately, but now
this instead happens directly when you call `addEntities` or `updateEntity`.
If you were using `addEntities` or `updateEntity` directly, please adapt your
code to add the `relations` array to each request. If you were calling
`setRelations` separately next to these methods, you no longer need to do so,
after adding the relations to the `DbEntityRequest`s.
plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.test.ts
+10 -12
View File
@@ -80,9 +80,10 @@ describe('DatabaseEntitiesCatalog', () => {
'kind=b,metadata.namespace=d,metadata.name=c',
),
);
expect(db.setRelations).toHaveBeenCalledTimes(1);
expect(db.setRelations).toHaveBeenCalledWith(expect.anything(), 'u', []);
expect(db.addEntities).toHaveBeenCalledTimes(1);
expect(db.addEntities).toHaveBeenCalledWith(expect.anything(), [
{ entity: expect.anything(), relations: [] },
]);
expect(result).toEqual([{ entityId: 'u' }]);
});
@@ -113,9 +114,10 @@ describe('DatabaseEntitiesCatalog', () => {
'kind=b,metadata.namespace=d,metadata.name=c',
),
);
expect(db.setRelations).toHaveBeenCalledTimes(1);
expect(db.setRelations).toHaveBeenCalledWith(expect.anything(), 'u', []);
expect(db.addEntities).toHaveBeenCalledTimes(1);
expect(db.addEntities).toHaveBeenCalledWith(expect.anything(), [
{ entity: expect.anything(), relations: [] },
]);
expect(transaction.rollback).toBeCalledTimes(1);
expect(result).toEqual([{ entityId: 'u' }]);
});
@@ -145,11 +147,7 @@ describe('DatabaseEntitiesCatalog', () => {
},
},
};
db.entities.mockResolvedValue([
{
entity: dbEntity,
},
]);
db.entities.mockResolvedValue([{ entity: dbEntity }]);
db.addEntities.mockResolvedValue([
{ entity: { ...entity, metadata: { ...entity.metadata, uid: 'u' } } },
]);
@@ -161,7 +159,7 @@ describe('DatabaseEntitiesCatalog', () => {
);
expect(db.entities).toHaveBeenCalledTimes(2);
expect(db.setRelations).toHaveBeenCalledTimes(1);
expect(db.addEntities).toHaveBeenCalledTimes(1);
expect(result).toEqual([
{
entityId: 'u',
@@ -237,12 +235,11 @@ describe('DatabaseEntitiesCatalog', () => {
x: 'b',
},
},
relations: [],
},
'e',
1,
);
expect(db.setRelations).toHaveBeenCalledTimes(1);
expect(db.setRelations).toHaveBeenCalledWith(expect.anything(), 'u', []);
expect(result).toEqual([{ entityId: 'u' }]);
});
@@ -315,6 +312,7 @@ describe('DatabaseEntitiesCatalog', () => {
x: 'b',
},
},
relations: [],
},
'e',
1,
plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.ts
+37 -50
View File
@@ -18,7 +18,6 @@ import { ConflictError, NotFoundError } from '@backstage/backend-common';
import {
Entity,
entityHasChanges,
EntityRelationSpec,
generateUpdatedEntity,
getEntityName,
LOCATION_ANNOTATION,
@@ -111,47 +110,45 @@ export class DatabaseEntitiesCatalog implements EntitiesCatalog {
outputEntities?: boolean;
},
): Promise<EntityUpsertResponse[]> {
// Group the entities by unique kind+namespace combinations. The reason for
// Group the requests by unique kind+namespace combinations. The reason for
// this is that the change detection and merging logic requires finding
// pre-existing versions of the entities in the database. Those queries are
// easier and faster to make if every batch revolves around a single kind-
// namespace pair.
const entitiesByKindAndNamespace = groupBy(requests, ({ entity }) => {
const requestsByKindAndNamespace = groupBy(requests, ({ entity }) => {
const name = getEntityName(entity);
return `${name.kind}:${name.namespace}`.toLowerCase();
});
// Go through the requests in reasonable batch sizes. Sometimes, sources
// produce tens of thousands of entities, and those are too large batch
// sizes to reasonably send to the database.
const batches = Object.values(requestsByKindAndNamespace)
.map(requests => chunk(requests, BATCH_SIZE))
.flat();
// Bound the number of concurrent batches. We want a bit of concurrency for
// performance reasons, but not so much that we starve the connection pool
// or start thrashing.
const limiter = limiterFactory(BATCH_CONCURRENCY);
const tasks = new Array<Promise<EntityUpsertResponse[]>>();
for (const groupRequests of Object.values(entitiesByKindAndNamespace)) {
// Go through the new entities in reasonable chunk sizes (sometimes,
// sources produce tens of thousands of entities, and those are too large
// batch sizes to reasonably send to the database)
for (const batch of chunk(groupRequests, BATCH_SIZE)) {
tasks.push(
limiter(async () => {
// Retry the batch write a few times to deal with contention
for (let attempt = 1; ; ++attempt) {
try {
return this.batchAddOrUpdateEntitiesSingleBatch(batch, options);
} catch (e) {
if (e instanceof ConflictError && attempt < BATCH_ATTEMPTS) {
this.logger.warn(
`Failed to write batch at attempt ${attempt}/${BATCH_ATTEMPTS}, ${e}`,
);
} else {
throw e;
}
}
const tasks = batches.map(batch =>
limiter(async () => {
// Retry the batch write a few times to deal with contention
for (let attempt = 1; ; ++attempt) {
try {
return this.batchAddOrUpdateEntitiesSingleBatch(batch, options);
} catch (e) {
if (e instanceof ConflictError && attempt < BATCH_ATTEMPTS) {
this.logger.warn(
`Failed to write batch at attempt ${attempt}/${BATCH_ATTEMPTS}, ${e}`,
);
} else {
throw e;
}
}),
);
}
}
}
}
}),
);
const responses = await Promise.all(tasks);
return responses.flat();
@@ -203,7 +200,7 @@ export class DatabaseEntitiesCatalog implements EntitiesCatalog {
// likely want to figure out a way to avoid that
const entityId = entity.metadata.uid;
if (entityId) {
await this.setRelations(entityId, relations, tx);
await this.database.setRelations(tx, entityId, relations);
responses.push({ entityId });
}
}
@@ -232,15 +229,6 @@ export class DatabaseEntitiesCatalog implements EntitiesCatalog {
});
}
// Set the relations originating from an entity using the DB layer
private async setRelations(
originatingEntityId: string,
relations: EntityRelationSpec[],
tx: Transaction,
): Promise<void> {
await this.database.setRelations(tx, originatingEntityId, relations);
}
// Given a batch of entities that were just read from a location, take them
// into consideration by comparing against the existing catalog entities and
// produce the list of entities to be added, and the list of entities to be
@@ -321,17 +309,17 @@ export class DatabaseEntitiesCatalog implements EntitiesCatalog {
const res = await this.database.addEntities(
tx,
requests.map(({ entity }) => ({ locationId, entity })),
requests.map(({ entity, relations }) => ({
locationId,
entity,
relations,
})),
);
const responses = res.map(({ entity }) => ({
entityId: entity.metadata.uid!,
}));
for (const [index, { entityId }] of responses.entries()) {
await this.setRelations(entityId, requests[index].relations, tx);
}
this.logger.debug(
`Added ${requests.length} entities in ${durationText(markTimestamp)}`,
);
@@ -350,11 +338,10 @@ export class DatabaseEntitiesCatalog implements EntitiesCatalog {
const responses: EntityUpsertResponse[] = [];
// TODO(freben): Still not batched
for (const entity of requests) {
const res = await this.addOrUpdateEntity(entity.entity, tx, locationId);
for (const request of requests) {
const res = await this.addOrUpdateEntity(tx, request, locationId);
const entityId = res.metadata.uid!;
responses.push({ entityId });
await this.setRelations(entityId, entity.relations, tx);
}
this.logger.debug(
@@ -366,8 +353,8 @@ export class DatabaseEntitiesCatalog implements EntitiesCatalog {
// TODO(freben): Incorporate this into batchUpdate which is the only caller
private async addOrUpdateEntity(
entity: Entity,
tx: Transaction,
{ entity, relations }: EntityUpsertRequest,
locationId?: string,
): Promise<Entity> {
// Find a matching (by uid, or by compound name, depending on the given
@@ -383,13 +370,13 @@ export class DatabaseEntitiesCatalog implements EntitiesCatalog {
const updated = generateUpdatedEntity(existing.entity, entity);
response = await this.database.updateEntity(
tx,
{ locationId, entity: updated },
{ locationId, entity: updated, relations },
existing.entity.metadata.etag,
existing.entity.metadata.generation,
);
} else {
const added = await this.database.addEntities(tx, [
{ locationId, entity },
{ locationId, entity, relations },
]);
response = added[0];
}
plugins/catalog-backend/src/database/CommonDatabase.test.ts
+29 -10
View File
@@ -54,6 +54,7 @@ describe('CommonDatabase', () => {
},
spec: { i: 'j' },
},
relations: [],
};
entityResponse = {
@@ -152,6 +153,7 @@ describe('CommonDatabase', () => {
kind: 'k1',
metadata: { name: 'n1', namespace: 'ns1' },
},
relations: [],
},
{
entity: {
@@ -159,6 +161,7 @@ describe('CommonDatabase', () => {
kind: 'k1',
metadata: { name: 'n1', namespace: 'ns1' },
},
relations: [],
},
];
await expect(
@@ -174,6 +177,7 @@ describe('CommonDatabase', () => {
kind: 'k1',
metadata: { name: 'n1', namespace: 'ns1' },
},
relations: [],
},
{
entity: {
@@ -181,6 +185,7 @@ describe('CommonDatabase', () => {
kind: 'k1',
metadata: { name: 'n1', namespace: 'nS1' },
},
relations: [],
},
];
await expect(
@@ -196,6 +201,7 @@ describe('CommonDatabase', () => {
kind: 'k1',
metadata: { name: 'n1', namespace: 'ns1' },
},
relations: [],
},
{
entity: {
@@ -203,6 +209,7 @@ describe('CommonDatabase', () => {
kind: 'k1',
metadata: { name: 'n1', namespace: 'ns2' },
},
relations: [],
},
];
await expect(
@@ -285,7 +292,7 @@ describe('CommonDatabase', () => {
db.addEntities(tx, [entityRequest]),
);
const updated = await db.transaction(tx =>
db.updateEntity(tx, { entity: added.entity }),
db.updateEntity(tx, { entity: added.entity, relations: [] }),
);
expect(updated.entity.apiVersion).toEqual(added.entity.apiVersion);
expect(updated.entity.kind).toEqual(added.entity.kind);
@@ -305,7 +312,7 @@ describe('CommonDatabase', () => {
);
added.entity.metadata.name! = 'new!';
const updated = await db.transaction(tx =>
db.updateEntity(tx, { entity: added.entity }),
db.updateEntity(tx, { entity: added.entity, relations: [] }),
);
expect(updated.entity.metadata.name).toEqual('new!');
});
@@ -316,7 +323,11 @@ describe('CommonDatabase', () => {
);
await expect(
db.transaction(tx =>
db.updateEntity(tx, { entity: added.entity }, 'garbage'),
db.updateEntity(
tx,
{ entity: added.entity, relations: [] },
'garbage',
),
),
).rejects.toThrow(ConflictError);
});
@@ -327,7 +338,12 @@ describe('CommonDatabase', () => {
);
await expect(
db.transaction(tx =>
db.updateEntity(tx, { entity: added.entity }, undefined, 1e20),
db.updateEntity(
tx,
{ entity: added.entity, relations: [] },
undefined,
1e20,
),
),
).rejects.toThrow(ConflictError);
});
@@ -347,7 +363,10 @@ describe('CommonDatabase', () => {
spec: { c: null },
};
await db.transaction(async tx => {
await db.addEntities(tx, [{ entity: e1 }, { entity: e2 }]);
await db.addEntities(tx, [
{ entity: e1, relations: [] },
{ entity: e2, relations: [] },
]);
});
const result = await db.transaction(async tx => db.entities(tx));
expect(result.length).toEqual(2);
@@ -385,7 +404,7 @@ describe('CommonDatabase', () => {
await db.transaction(async tx => {
await db.addEntities(
tx,
entities.map(entity => ({ entity })),
entities.map(entity => ({ entity, relations: [] })),
);
});
@@ -426,7 +445,7 @@ describe('CommonDatabase', () => {
await db.transaction(async tx => {
await db.addEntities(
tx,
entities.map(entity => ({ entity })),
entities.map(entity => ({ entity, relations: [] })),
);
});
@@ -566,8 +585,8 @@ describe('CommonDatabase', () => {
const { id2: secondEntityId } = await db.transaction(async tx => {
const [{ entity: e1 }, { entity: e2 }] = await db.addEntities(tx, [
{ entity: entity1 },
{ entity: entity2 },
{ entity: entity1, relations: [] },
{ entity: entity2, relations: [] },
]);
const id1 = e1?.metadata?.uid!;
const id2 = e2?.metadata?.uid!;
@@ -665,7 +684,7 @@ describe('CommonDatabase', () => {
await db.transaction(async tx => {
await db.addEntities(
tx,
entities.map(entity => ({ entity })),
entities.map(entity => ({ entity, relations: [] })),
);
});
plugins/catalog-backend/src/database/CommonDatabase.ts
+60 -100
View File
@@ -95,35 +95,6 @@ export class CommonDatabase implements Database {
}
}
async addEntity(
txOpaque: Transaction,
request: DbEntityRequest,
): Promise<DbEntityResponse> {
const tx = txOpaque as Knex.Transaction<any, any>;
if (request.entity.metadata.uid !== undefined) {
throw new InputError('May not specify uid for new entities');
} else if (request.entity.metadata.etag !== undefined) {
throw new InputError('May not specify etag for new entities');
} else if (request.entity.metadata.generation !== undefined) {
throw new InputError('May not specify generation for new entities');
}
const newEntity = lodash.cloneDeep(request.entity);
newEntity.metadata = {
...newEntity.metadata,
uid: generateEntityUid(),
etag: generateEntityEtag(),
generation: 1,
};
const newRow = this.toEntityRow(request.locationId, newEntity);
await tx<DbEntitiesRow>('entities').insert(newRow);
await this.updateEntitiesSearch(tx, newRow.id, newEntity);
return { locationId: request.locationId, entity: newEntity };
}
async addEntities(
txOpaque: Transaction,
request: DbEntityRequest[],
@@ -132,9 +103,10 @@ export class CommonDatabase implements Database {
const result: DbEntityResponse[] = [];
const entityRows: DbEntitiesRow[] = [];
const relationRows: DbEntitiesRelationsRow[] = [];
const searchRows: DbEntitiesSearchRow[] = [];
for (const { entity, locationId } of request) {
for (const { entity, relations, locationId } of request) {
if (entity.metadata.uid !== undefined) {
throw new InputError('May not specify uid for new entities');
} else if (entity.metadata.etag !== undefined) {
@@ -145,28 +117,27 @@ export class CommonDatabase implements Database {
throw new InputError('May not specify relations for new entities');
}
const uid = generateEntityUid();
const etag = generateEntityEtag();
const generation = 1;
const newEntity = {
...entity,
metadata: {
...entity.metadata,
uid: generateEntityUid(),
etag: generateEntityEtag(),
generation: 1,
uid,
etag,
generation,
},
};
result.push({ entity: newEntity, locationId });
entityRows.push(this.toEntityRow(locationId, newEntity));
searchRows.push(...buildEntitySearch(newEntity.metadata.uid, newEntity));
relationRows.push(...this.toRelationRows(uid, relations));
searchRows.push(...buildEntitySearch(uid, newEntity));
}
await tx.batchInsert('entities', entityRows, BATCH_SIZE);
await tx<DbEntitiesSearchRow>('entities_search')
.whereIn(
'entity_id',
entityRows.map(r => r.id),
)
.del();
await tx.batchInsert('entities_relations', relationRows, BATCH_SIZE);
await tx.batchInsert('entities_search', searchRows, BATCH_SIZE);
return result;
@@ -181,8 +152,7 @@ export class CommonDatabase implements Database {
const tx = txOpaque as Knex.Transaction<any, any>;
const { uid } = request.entity.metadata;
if (uid === undefined) {
if (!uid) {
throw new InputError('Must specify uid when updating entities');
}
@@ -193,40 +163,48 @@ export class CommonDatabase implements Database {
if (oldRows.length !== 1) {
throw new NotFoundError('No matching entity found');
}
const etag = oldRows[0].etag;
const generation = Number(oldRows[0].generation);
// Validate the old entity
const oldRow = oldRows[0];
// The Number cast is here because sqlite reads it as a string, no matter
// what the table actually says
oldRow.generation = Number(oldRow.generation);
if (matchingEtag) {
if (matchingEtag !== oldRow.etag) {
throw new ConflictError(
`Etag mismatch, expected="${matchingEtag}" found="${oldRow.etag}"`,
);
}
// Validate the old entity. The Number cast is here because sqlite reads it
// as a string, no matter what the table actually says.
if (matchingEtag && matchingEtag !== etag) {
throw new ConflictError(
`Etag mismatch, expected="${matchingEtag}" found="${etag}"`,
);
}
if (matchingGeneration) {
if (matchingGeneration !== oldRow.generation) {
throw new ConflictError(
`Generation mismatch, expected="${matchingGeneration}" found="${oldRow.generation}"`,
);
}
if (matchingGeneration && matchingGeneration !== generation) {
throw new ConflictError(
`Generation mismatch, expected="${matchingGeneration}" found="${generation}"`,
);
}
// Store the updated entity; select on the old etag to ensure that we do
// not lose to another writer
const newRow = this.toEntityRow(request.locationId, request.entity);
const updatedRows = await tx<DbEntitiesRow>('entities')
.where({ id: oldRow.id, etag: oldRow.etag })
.where({ id: uid, etag })
.update(newRow);
// If this happens, somebody else changed the entity just now
if (updatedRows !== 1) {
throw new ConflictError(`Failed to update entity`);
}
await this.updateEntitiesSearch(tx, oldRow.id, request.entity);
const relationRows = this.toRelationRows(uid, request.relations);
await tx<DbEntitiesRelationsRow>('entities_relations')
.where({ originating_entity_id: uid })
.del();
await tx.batchInsert('entities_relations', relationRows, BATCH_SIZE);
try {
const entries = buildEntitySearch(uid, request.entity);
await tx<DbEntitiesSearchRow>('entities_search')
.where({ entity_id: uid })
.del();
await tx.batchInsert('entities_search', entries, BATCH_SIZE);
} catch {
// ignore intentionally - if this happens, the entity was deleted before
// we got around to writing the entries
}
return request;
}
@@ -326,30 +304,12 @@ export class CommonDatabase implements Database {
relations: EntityRelationSpec[],
): Promise<void> {
const tx = txOpaque as Knex.Transaction<any, any>;
const relationRows = this.toRelationRows(originatingEntityId, relations);
// remove all relations that exist for the originating entity id.
await tx<DbEntitiesRelationsRow>('entities_relations')
.where({ originating_entity_id: originatingEntityId })
.del();
const serializeName = (e: EntityName) =>
`${e.kind}:${e.namespace}/${e.name}`.toLowerCase();
const relationsRows: DbEntitiesRelationsRow[] = relations.map(
({ source, target, type }) => ({
originating_entity_id: originatingEntityId,
source_full_name: serializeName(source),
target_full_name: serializeName(target),
type,
}),
);
// TODO(blam): translate constraint failures to sane NotFoundError instead
await tx.batchInsert(
'entities_relations',
deduplicateRelations(relationsRows),
BATCH_SIZE,
);
await tx.batchInsert('entities_relations', relationRows, BATCH_SIZE);
}
async addLocation(
@@ -458,23 +418,6 @@ export class CommonDatabase implements Database {
}
}
private async updateEntitiesSearch(
tx: Knex.Transaction<any, any>,
entityId: string,
data: Entity,
): Promise<void> {
try {
const entries = buildEntitySearch(entityId, data);
await tx<DbEntitiesSearchRow>('entities_search')
.where({ entity_id: entityId })
.del();
await tx<DbEntitiesSearchRow>('entities_search').insert(entries);
} catch {
// ignore intentionally - if this happens, the entity was deleted before
// we got around to writing the entries
}
}
private toEntityRow(
locationId: string | undefined,
entity: Entity,
@@ -500,6 +443,23 @@ export class CommonDatabase implements Database {
};
}
private toRelationRows(
originatingEntityId: string,
relations: EntityRelationSpec[],
): DbEntitiesRelationsRow[] {
const serializeName = (e: EntityName) =>
`${e.kind}:${e.namespace}/${e.name}`.toLowerCase();
const rows = relations.map(({ source, target, type }) => ({
originating_entity_id: originatingEntityId,
source_full_name: serializeName(source),
target_full_name: serializeName(target),
type,
}));
return deduplicateRelations(rows);
}
private async toEntityResponses(
tx: Knex.Transaction<any, any>,
rows: DbEntitiesRow[],
plugins/catalog-backend/src/database/types.ts
+6 -3
View File
@@ -33,6 +33,7 @@ export type DbEntitiesRow = {
export type DbEntityRequest = {
locationId?: string;
entity: Entity;
relations: EntityRelationSpec[];
};
export type DbEntityResponse = {
@@ -184,10 +185,12 @@ export type Database = {
removeEntityByUid(tx: Transaction, uid: string): Promise<void>;
/**
* Remove current relations for the entity and replace them with the new relations array
* Remove current relations for the entity and replace them with the new
* relations array.
*
* @param tx An ongoing transaction
* @param entityUid the entity uid
* @param relations the relationships to be set
* @param entityUid The entity uid
* @param relations The relationships to be set
*/
setRelations(
tx: Transaction,