diff --git a/.changeset/purple-cycles-switch.md b/.changeset/purple-cycles-switch.md new file mode 100644 index 0000000000..c7594e4254 --- /dev/null +++ b/.changeset/purple-cycles-switch.md @@ -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. diff --git a/plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.test.ts b/plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.test.ts index 8d2138b3e5..5059d741f4 100644 --- a/plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.test.ts +++ b/plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.test.ts @@ -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, diff --git a/plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.ts b/plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.ts index 84234d85a2..7ba6897236 100644 --- a/plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.ts +++ b/plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.ts @@ -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 { - // 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>(); - - 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 { - 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 { // 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]; } diff --git a/plugins/catalog-backend/src/database/CommonDatabase.test.ts b/plugins/catalog-backend/src/database/CommonDatabase.test.ts index ffffd85f04..130ed41eb1 100644 --- a/plugins/catalog-backend/src/database/CommonDatabase.test.ts +++ b/plugins/catalog-backend/src/database/CommonDatabase.test.ts @@ -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: [] })), ); }); diff --git a/plugins/catalog-backend/src/database/CommonDatabase.ts b/plugins/catalog-backend/src/database/CommonDatabase.ts index f4c3e75934..d641cbb7b6 100644 --- a/plugins/catalog-backend/src/database/CommonDatabase.ts +++ b/plugins/catalog-backend/src/database/CommonDatabase.ts @@ -95,35 +95,6 @@ export class CommonDatabase implements Database { } } - async addEntity( - txOpaque: Transaction, - request: DbEntityRequest, - ): Promise { - const tx = txOpaque as Knex.Transaction; - - 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('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('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; 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('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('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('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 { const tx = txOpaque as Knex.Transaction; + const relationRows = this.toRelationRows(originatingEntityId, relations); - // remove all relations that exist for the originating entity id. await tx('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, - entityId: string, - data: Entity, - ): Promise { - try { - const entries = buildEntitySearch(entityId, data); - await tx('entities_search') - .where({ entity_id: entityId }) - .del(); - await tx('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, rows: DbEntitiesRow[], diff --git a/plugins/catalog-backend/src/database/types.ts b/plugins/catalog-backend/src/database/types.ts index 320ec1ef18..5a9f60853a 100644 --- a/plugins/catalog-backend/src/database/types.ts +++ b/plugins/catalog-backend/src/database/types.ts @@ -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; /** - * 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,