diff --git a/packages/catalog-model/src/kinds/GroupEntityV1alpha1.ts b/packages/catalog-model/src/kinds/GroupEntityV1alpha1.ts index f0e91ddce8..8d1fbe78e8 100644 --- a/packages/catalog-model/src/kinds/GroupEntityV1alpha1.ts +++ b/packages/catalog-model/src/kinds/GroupEntityV1alpha1.ts @@ -32,21 +32,21 @@ const schema = yup.object>({ // one element and there is no simple workaround -_- // the cast is there to convince typescript that the array itself is // required without using .required() - ancestors: yup.array(yup.string().required()).test({ - name: 'isDefined', - message: 'ancestors must be defined', - test: v => Boolean(v), - }) as yup.ArraySchema, - children: yup.array(yup.string().required()).test({ - name: 'isDefined', - message: 'children must be defined', - test: v => Boolean(v), - }) as yup.ArraySchema, - descendants: yup.array(yup.string().required()).test({ - name: 'isDefined', - message: 'descendants must be defined', - test: v => Boolean(v), - }) as yup.ArraySchema, + // ancestors: yup.array(yup.string().required()).test({ + // name: 'isDefined', + // message: 'ancestors must be defined', + // test: v => Boolean(v), + // }) as yup.ArraySchema, + // children: yup.array(yup.string().required()).test({ + // name: 'isDefined', + // message: 'children must be defined', + // test: v => Boolean(v), + // }) as yup.ArraySchema, + // descendants: yup.array(yup.string().required()).test({ + // name: 'isDefined', + // message: 'descendants must be defined', + // test: v => Boolean(v), + // }) as yup.ArraySchema, }) .required(), }); diff --git a/packages/catalog-model/src/kinds/UserEntityV1alpha1.ts b/packages/catalog-model/src/kinds/UserEntityV1alpha1.ts index 6086b3776e..293f87b319 100644 --- a/packages/catalog-model/src/kinds/UserEntityV1alpha1.ts +++ b/packages/catalog-model/src/kinds/UserEntityV1alpha1.ts @@ -37,11 +37,11 @@ const schema = yup.object>({ // element and there is no simple workaround -_- // the cast is there to convince typescript that the array itself is // required without using .required() - memberOf: yup.array(yup.string().required()).test({ - name: 'isDefined', - message: 'memberOf must be defined', - test: v => Boolean(v), - }) as yup.ArraySchema, + // memberOf: yup.array(yup.string().required()).test({ + // name: 'isDefined', + // message: 'memberOf must be defined', + // test: v => Boolean(v), + // }) as yup.ArraySchema, }) .required(), }); diff --git a/plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.ts b/plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.ts index efb541727a..e4accf9a8d 100644 --- a/plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.ts +++ b/plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.ts @@ -29,7 +29,11 @@ import limiterFactory from 'p-limit'; import { Logger } from 'winston'; import type { Database, DbEntityResponse, EntityFilters } from '../database'; import { durationText } from '../util/timing'; -import type { EntitiesCatalog } from './types'; +import type { + EntitiesCatalog, + EntityMutationRequest, + EntityMutationResponse, +} from './types'; type BatchContext = { kind: string; @@ -63,7 +67,7 @@ export class DatabaseEntitiesCatalog implements EntitiesCatalog { return items.map(i => i.entity); } - async addOrUpdateEntity( + private async addOrUpdateEntity( entity: Entity, locationId?: string, ): Promise { @@ -96,15 +100,6 @@ export class DatabaseEntitiesCatalog implements EntitiesCatalog { }); } - async addEntities(entities: Entity[], locationId?: string): Promise { - await this.database.transaction(async tx => { - await this.database.addEntities( - tx, - entities.map(entity => ({ locationId, entity })), - ); - }); - } - async removeEntityByUid(uid: string): Promise { return await this.database.transaction(async tx => { const entityResponse = await this.database.entityByUid(tx, uid); @@ -138,27 +133,31 @@ export class DatabaseEntitiesCatalog implements EntitiesCatalog { * @param entities Some entities * @param locationId The location that they all belong to */ - async batchAddOrUpdateEntities(entities: Entity[], locationId?: string) { + async batchAddOrUpdateEntities( + requests: EntityMutationRequest[], + locationId?: string, + ): Promise { // Group the entities by unique kind+namespace combinations - const entitiesByKindAndNamespace = groupBy(entities, entity => { + const entitiesByKindAndNamespace = groupBy(requests, ({ entity }) => { const name = getEntityName(entity); return `${name.kind}:${name.namespace}`.toLowerCase(); }); const limiter = limiterFactory(BATCH_CONCURRENCY); - const tasks: Promise[] = []; + const tasks: Promise[] = []; - for (const groupEntities of Object.values(entitiesByKindAndNamespace)) { - const { kind, namespace } = getEntityName(groupEntities[0]); + for (const groupRequests of Object.values(entitiesByKindAndNamespace)) { + const { kind, namespace } = getEntityName(groupRequests[0].entity); // 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(groupEntities, BATCH_SIZE)) { + for (const batch of chunk(groupRequests, BATCH_SIZE)) { tasks.push( limiter(async () => { - const first = serializeEntityRef(batch[0]); - const last = serializeEntityRef(batch[batch.length - 1]); + const first = serializeEntityRef(batch[0].entity); + const last = serializeEntityRef(batch[batch.length - 1].entity); + const modifiedEntityIds: EntityMutationResponse[] = []; this.logger.debug( `Considering batch ${first}-${last} (${batch.length} entries)`, ); @@ -171,8 +170,16 @@ export class DatabaseEntitiesCatalog implements EntitiesCatalog { batch, context, ); - if (toAdd.length) await this.batchAdd(toAdd, context); - if (toUpdate.length) await this.batchUpdate(toUpdate, context); + if (toAdd.length) { + modifiedEntityIds.push( + ...(await this.batchAdd(toAdd, context)), + ); + } + if (toUpdate.length) { + modifiedEntityIds.push( + ...(await this.batchUpdate(toUpdate, context)), + ); + } break; } catch (e) { if (e instanceof ConflictError && attempt < BATCH_ATTEMPTS) { @@ -184,16 +191,18 @@ export class DatabaseEntitiesCatalog implements EntitiesCatalog { } } } + + return modifiedEntityIds; }), ); } } - await Promise.all(tasks); + return (await Promise.all(tasks)).flat(); } // Set the relations originating from an entity using the DB layer - async setRelations( + private async setRelations( originatingEntityId: string, relations: EntityRelationSpec[], ): Promise { @@ -207,15 +216,15 @@ export class DatabaseEntitiesCatalog implements EntitiesCatalog { // produce the list of entities to be added, and the list of entities to be // updated private async analyzeBatch( - newEntities: Entity[], + requests: EntityMutationRequest[], { kind, namespace }: BatchContext, ): Promise<{ - toAdd: Entity[]; - toUpdate: Entity[]; + toAdd: EntityMutationRequest[]; + toUpdate: EntityMutationRequest[]; }> { const markTimestamp = process.hrtime(); - const names = newEntities.map(e => e.metadata.name); + const names = requests.map(({ entity }) => entity.metadata.name); const oldEntities = await this.entities({ kind: kind, 'metadata.namespace': namespace, @@ -226,18 +235,19 @@ export class DatabaseEntitiesCatalog implements EntitiesCatalog { oldEntities.map(e => [e.metadata.name, e]), ); - const toAdd: Entity[] = []; - const toUpdate: Entity[] = []; + const toAdd: EntityMutationRequest[] = []; + const toUpdate: EntityMutationRequest[] = []; - for (const newEntity of newEntities) { + for (const request of requests) { + const newEntity = request.entity; const oldEntity = oldEntitiesByName.get(newEntity.metadata.name); if (!oldEntity) { - toAdd.push(newEntity); + toAdd.push(request); } else if (entityHasChanges(oldEntity, newEntity)) { // TODO(freben): This currently uses addOrUpdateEntity under the hood, // but should probably calculate the end result entity right here // instead and call a dedicated batch update database method instead - toUpdate.push(newEntity); + toUpdate.push(request); } } @@ -252,28 +262,55 @@ export class DatabaseEntitiesCatalog implements EntitiesCatalog { // Efficiently adds the given entities to storage, under the assumption that // they do not conflict with any existing entities - private async batchAdd(entities: Entity[], { locationId }: BatchContext) { + private async batchAdd( + requests: EntityMutationRequest[], + { locationId }: BatchContext, + ): Promise { const markTimestamp = process.hrtime(); - await this.addEntities(entities, locationId); + const res = await this.database.transaction( + async tx => + await this.database.addEntities( + tx, + requests.map(({ entity }) => ({ locationId, entity })), + ), + ); + + const entityIds = res.map(({ entity }) => ({ + entityId: entity.metadata.uid!, + })); + + for (const [index, { entityId }] of entityIds.entries()) { + await this.setRelations(entityId, requests[index].relations); + } this.logger.debug( - `Added ${entities.length} entities in ${durationText(markTimestamp)}`, + `Added ${requests.length} entities in ${durationText(markTimestamp)}`, ); + + return entityIds; } // Efficiently updates the given entities into storage, under the assumption // that there already exist entities with the same names - private async batchUpdate(entities: Entity[], { locationId }: BatchContext) { + private async batchUpdate( + requests: EntityMutationRequest[], + { locationId }: BatchContext, + ): Promise { const markTimestamp = process.hrtime(); - + const responseIds: EntityMutationResponse[] = []; // TODO(freben): Still not batched - for (const entity of entities) { - await this.addOrUpdateEntity(entity, locationId); + for (const entity of requests) { + const res = await this.addOrUpdateEntity(entity.entity, locationId); + const entityId = res.metadata.uid!; + responseIds.push({ entityId }); + await this.setRelations(entityId, entity.relations); } this.logger.debug( - `Updated ${entities.length} entities in ${durationText(markTimestamp)}`, + `Updated ${requests.length} entities in ${durationText(markTimestamp)}`, ); + + return responseIds; } } diff --git a/plugins/catalog-backend/src/catalog/types.ts b/plugins/catalog-backend/src/catalog/types.ts index e50b5f15db..c9c514c42f 100644 --- a/plugins/catalog-backend/src/catalog/types.ts +++ b/plugins/catalog-backend/src/catalog/types.ts @@ -21,10 +21,17 @@ import type { EntityFilters } from '../database'; // Entities // +export type EntityMutationRequest = { + entity: Entity; + relations: EntityRelationSpec[]; +}; + +export type EntityMutationResponse = { + entityId: string; +}; + export type EntitiesCatalog = { entities(filters?: EntityFilters): Promise; - addOrUpdateEntity(entity: Entity, locationId?: string): Promise; - addEntities(entities: Entity[], locationId?: string): Promise; removeEntityByUid(uid: string): Promise; /** @@ -34,15 +41,9 @@ export type EntitiesCatalog = { * @param locationId The location that they all belong to */ batchAddOrUpdateEntities( - entities: Entity[], + entities: EntityMutationRequest[], locationId?: string, - ): Promise; - - // Same as the DB layer - setRelations( - entityUid: string, - relations: EntityRelationSpec[], - ): Promise; + ): Promise; }; // diff --git a/plugins/catalog-backend/src/ingestion/HigherOrderOperations.test.ts b/plugins/catalog-backend/src/ingestion/HigherOrderOperations.test.ts index c7c8e738c8..b8aaa5fb13 100644 --- a/plugins/catalog-backend/src/ingestion/HigherOrderOperations.test.ts +++ b/plugins/catalog-backend/src/ingestion/HigherOrderOperations.test.ts @@ -31,10 +31,7 @@ describe('HigherOrderOperations', () => { beforeAll(() => { entitiesCatalog = { entities: jest.fn(), - addOrUpdateEntity: jest.fn(), - addEntities: jest.fn(), removeEntityByUid: jest.fn(), - setRelations: jest.fn(), batchAddOrUpdateEntities: jest.fn(), }; locationsCatalog = { @@ -72,7 +69,6 @@ describe('HigherOrderOperations', () => { locationReader.read.mockResolvedValue({ entities: [], errors: [], - relations: [], }); const result = await higherOrderOperation.addLocation(spec); @@ -116,7 +112,6 @@ describe('HigherOrderOperations', () => { locationReader.read.mockResolvedValue({ entities: [], errors: [], - relations: [], }); const result = await higherOrderOperation.addLocation(spec); @@ -144,9 +139,8 @@ describe('HigherOrderOperations', () => { locationsCatalog.locations.mockResolvedValue([]); locationReader.read.mockResolvedValue({ - entities: [{ entity, location }], + entities: [{ entity, location, relations: [] }], errors: [{ error: new Error('abcd'), location }], - relations: [], }); await expect(higherOrderOperation.addLocation(spec)).rejects.toThrow( @@ -193,9 +187,8 @@ describe('HigherOrderOperations', () => { { currentStatus: locationStatus, data: location }, ]); locationReader.read.mockResolvedValue({ - entities: [{ entity: desc, location }], + entities: [{ entity: desc, location, relations: [] }], errors: [], - relations: [], }); entitiesCatalog.batchAddOrUpdateEntities.mockResolvedValue(undefined); @@ -239,9 +232,8 @@ describe('HigherOrderOperations', () => { { currentStatus: locationStatus, data: location }, ]); locationReader.read.mockResolvedValue({ - entities: [{ entity: desc, location }], + entities: [{ entity: desc, location, relations: [] }], errors: [], - relations: [], }); entitiesCatalog.entities.mockResolvedValue([]); entitiesCatalog.addEntities.mockResolvedValue(undefined); diff --git a/plugins/catalog-backend/src/ingestion/HigherOrderOperations.ts b/plugins/catalog-backend/src/ingestion/HigherOrderOperations.ts index 6df6d9b4b7..2771945db2 100644 --- a/plugins/catalog-backend/src/ingestion/HigherOrderOperations.ts +++ b/plugins/catalog-backend/src/ingestion/HigherOrderOperations.ts @@ -15,7 +15,7 @@ */ import { InputError } from '@backstage/backend-common'; -import { Entity, Location, LocationSpec } from '@backstage/catalog-model'; +import { Location, LocationSpec } from '@backstage/catalog-model'; import { v4 as uuidv4 } from 'uuid'; import { Logger } from 'winston'; import { EntitiesCatalog, LocationsCatalog } from '../catalog'; @@ -94,16 +94,16 @@ export class HigherOrderOperations implements HigherOrderOperation { if (!previousLocation) { await this.locationsCatalog.addLocation(location); } - const outputEntities: Entity[] = []; - for (const entity of readerOutput.entities) { - const out = await this.entitiesCatalog.addOrUpdateEntity( - entity.entity, - location.id, - ); - outputEntities.push(out); - } + const writtenEntities = await this.entitiesCatalog.batchAddOrUpdateEntities( + readerOutput.entities, + location.id, + ); - return { location, entities: outputEntities }; + const entities = await this.entitiesCatalog.entities({ + 'metadata.uid': writtenEntities.map(e => e.entityId), + }); + + return { location, entities }; } /** @@ -166,7 +166,7 @@ export class HigherOrderOperations implements HigherOrderOperation { try { await this.entitiesCatalog.batchAddOrUpdateEntities( - readerOutput.entities.map(e => e.entity), + readerOutput.entities, location.id, ); } catch (e) { diff --git a/plugins/catalog-backend/src/ingestion/LocationReaders.ts b/plugins/catalog-backend/src/ingestion/LocationReaders.ts index 597a8ad6a3..215bfc9dc1 100644 --- a/plugins/catalog-backend/src/ingestion/LocationReaders.ts +++ b/plugins/catalog-backend/src/ingestion/LocationReaders.ts @@ -18,6 +18,7 @@ import { UrlReader } from '@backstage/backend-common'; import { Entity, EntityPolicy, + EntityRelationSpec, ENTITY_DEFAULT_NAMESPACE, LocationSpec, } from '@backstage/catalog-model'; @@ -64,7 +65,6 @@ export class LocationReaders implements LocationReader { const output: ReadLocationResult = { entities: [], errors: [], - relations: [], }; let items: CatalogProcessorResult[] = [result.location(location, false)]; @@ -79,11 +79,21 @@ export class LocationReaders implements LocationReader { await this.handleData(item, emit); } else if (item.type === 'entity') { if (rulesEnforcer.isAllowed(item.entity, item.location)) { - const entity = await this.handleEntity(item, emit); + const relations = Array(); + + const entity = await this.handleEntity(item, emitResult => { + if (emitResult.type === 'relation') { + relations.push(emitResult.relation); + return; + } + emit(emitResult); + }); + if (entity) { output.entities.push({ entity, location: item.location, + relations, }); } } else { @@ -100,10 +110,6 @@ export class LocationReaders implements LocationReader { location: item.location, error: item.error, }); - } else if (item.type === 'relation') { - output.relations.push({ - relation: item.relation, - }); } } @@ -126,11 +132,23 @@ export class LocationReaders implements LocationReader { ) { const { processors, logger } = this.options; + const validatedEmit: CatalogProcessorEmit = emitResult => { + if (emitResult.type === 'relation') { + throw new Error('readLocation may not emit entity relations'); + } + + emit(emitResult); + }; + for (const processor of processors) { if (processor.readLocation) { try { if ( - await processor.readLocation(item.location, item.optional, emit) + await processor.readLocation( + item.location, + item.optional, + validatedEmit, + ) ) { return; } @@ -153,10 +171,20 @@ export class LocationReaders implements LocationReader { ) { const { processors, logger } = this.options; + const validatedEmit: CatalogProcessorEmit = emitResult => { + if (emitResult.type === 'relation') { + throw new Error('parseData may not emit entity relations'); + } + + emit(emitResult); + }; + for (const processor of processors) { if (processor.parseData) { try { - if (await processor.parseData(item.data, item.location, emit)) { + if ( + await processor.parseData(item.data, item.location, validatedEmit) + ) { return; } } catch (e) { @@ -250,10 +278,18 @@ export class LocationReaders implements LocationReader { `Encountered error at location ${item.location.type} ${item.location.target}, ${item.error}`, ); + const validatedEmit: CatalogProcessorEmit = emitResult => { + if (emitResult.type === 'relation') { + throw new Error('handleError may not emit entity relations'); + } + + emit(emitResult); + }; + for (const processor of processors) { if (processor.handleError) { try { - await processor.handleError(item.error, item.location, emit); + await processor.handleError(item.error, item.location, validatedEmit); } catch (e) { const message = `Processor ${processor.constructor.name} threw an error while handling another error at ${item.location.type} ${item.location.target}, ${e}`; emit(result.generalError(item.location, message)); diff --git a/plugins/catalog-backend/src/ingestion/processors/types.ts b/plugins/catalog-backend/src/ingestion/processors/types.ts index 9ed5e6e92d..7821a61437 100644 --- a/plugins/catalog-backend/src/ingestion/processors/types.ts +++ b/plugins/catalog-backend/src/ingestion/processors/types.ts @@ -120,6 +120,7 @@ export type CatalogProcessorEntityResult = { export type CatalogProcessorRelationResult = { type: 'relation'; relation: EntityRelationSpec; + entityRef?: string; }; export type CatalogProcessorErrorResult = { diff --git a/plugins/catalog-backend/src/ingestion/types.ts b/plugins/catalog-backend/src/ingestion/types.ts index acaae61cc9..c586e2d438 100644 --- a/plugins/catalog-backend/src/ingestion/types.ts +++ b/plugins/catalog-backend/src/ingestion/types.ts @@ -52,13 +52,13 @@ export type LocationReader = { export type ReadLocationResult = { entities: ReadLocationEntity[]; - relations: { relation: EntityRelationSpec }[]; errors: ReadLocationError[]; }; export type ReadLocationEntity = { location: LocationSpec; entity: Entity; + relations: EntityRelationSpec[]; }; export type ReadLocationError = { diff --git a/plugins/catalog-backend/src/service/router.test.ts b/plugins/catalog-backend/src/service/router.test.ts index a2b17ba360..71caeb3d6c 100644 --- a/plugins/catalog-backend/src/service/router.test.ts +++ b/plugins/catalog-backend/src/service/router.test.ts @@ -32,10 +32,7 @@ describe('createRouter', () => { beforeAll(async () => { entitiesCatalog = { entities: jest.fn(), - addOrUpdateEntity: jest.fn(), - addEntities: jest.fn(), removeEntityByUid: jest.fn(), - setRelations: jest.fn(), batchAddOrUpdateEntities: jest.fn(), }; locationsCatalog = { diff --git a/plugins/catalog-backend/src/service/router.ts b/plugins/catalog-backend/src/service/router.ts index a262e9b5f4..5af773f1a0 100644 --- a/plugins/catalog-backend/src/service/router.ts +++ b/plugins/catalog-backend/src/service/router.ts @@ -49,8 +49,13 @@ export async function createRouter( }) .post('/entities', async (req, res) => { const body = await requireRequestBody(req); - const result = await entitiesCatalog.addOrUpdateEntity(body as Entity); - res.status(200).send(result); + const [result] = await entitiesCatalog.batchAddOrUpdateEntities([ + { entity: body as Entity, relations: [] }, + ]); + const entity = await entitiesCatalog.entities({ + 'metadata.uid': result.entityId, + }); + res.status(200).send(entity); }) .get('/entities/by-uid/:uid', async (req, res) => { const { uid } = req.params;