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

catalog-backend: relations working e2e

Browse Source 
Co-authored-by: blam <ben@blam.sh>
This commit is contained in:
Patrik Oldsberg
2020-10-20 17:57:11 +02:00
parent 62493e54c1
commit e26201d9d2
11 changed files with 176 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/catalog-model/src/kinds/GroupEntityV1alpha1.ts
+15 -15
View File
@@ -32,21 +32,21 @@ const schema = yup.object<Partial<GroupEntityV1alpha1>>({
// 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<string, object>,
children: yup.array(yup.string().required()).test({
name: 'isDefined',
message: 'children must be defined',
test: v => Boolean(v),
}) as yup.ArraySchema<string, object>,
descendants: yup.array(yup.string().required()).test({
name: 'isDefined',
message: 'descendants must be defined',
test: v => Boolean(v),
}) as yup.ArraySchema<string, object>,
// ancestors: yup.array(yup.string().required()).test({
// name: 'isDefined',
// message: 'ancestors must be defined',
// test: v => Boolean(v),
// }) as yup.ArraySchema<string, object>,
// children: yup.array(yup.string().required()).test({
// name: 'isDefined',
// message: 'children must be defined',
// test: v => Boolean(v),
// }) as yup.ArraySchema<string, object>,
// descendants: yup.array(yup.string().required()).test({
// name: 'isDefined',
// message: 'descendants must be defined',
// test: v => Boolean(v),
// }) as yup.ArraySchema<string, object>,
})
.required(),
});
packages/catalog-model/src/kinds/UserEntityV1alpha1.ts
+5 -5
View File
@@ -37,11 +37,11 @@ const schema = yup.object<Partial<UserEntityV1alpha1>>({
// 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<string, object>,
// memberOf: yup.array(yup.string().required()).test({
// name: 'isDefined',
// message: 'memberOf must be defined',
// test: v => Boolean(v),
// }) as yup.ArraySchema<string, object>,
})
.required(),
});
plugins/catalog-backend/src/catalog/DatabaseEntitiesCatalog.ts
+77 -40
View File
@@ -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<Entity> {
@@ -96,15 +100,6 @@ export class DatabaseEntitiesCatalog implements EntitiesCatalog {
});
}
async addEntities(entities: Entity[], locationId?: string): Promise<void> {
await this.database.transaction(async tx => {
await this.database.addEntities(
tx,
entities.map(entity => ({ locationId, entity })),
);
});
}
async removeEntityByUid(uid: string): Promise<void> {
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<EntityMutationResponse[]> {
// 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<void>[] = [];
const tasks: Promise<EntityMutationResponse[]>[] = [];
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<void> {
@@ -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<EntityMutationResponse[]> {
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<EntityMutationResponse[]> {
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;
}
}
plugins/catalog-backend/src/catalog/types.ts
+11 -10
View File
@@ -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<Entity[]>;
addOrUpdateEntity(entity: Entity, locationId?: string): Promise<Entity>;
addEntities(entities: Entity[], locationId?: string): Promise<void>;
removeEntityByUid(uid: string): Promise<void>;
/**
@@ -34,15 +41,9 @@ export type EntitiesCatalog = {
* @param locationId The location that they all belong to
*/
batchAddOrUpdateEntities(
entities: Entity[],
entities: EntityMutationRequest[],
locationId?: string,
): Promise<void>;
// Same as the DB layer
setRelations(
entityUid: string,
relations: EntityRelationSpec[],
): Promise<void>;
): Promise<EntityMutationResponse[]>;
};
//
plugins/catalog-backend/src/ingestion/HigherOrderOperations.test.ts
+3 -11
View File
@@ -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);
plugins/catalog-backend/src/ingestion/HigherOrderOperations.ts
+11 -11
View File
@@ -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) {
plugins/catalog-backend/src/ingestion/LocationReaders.ts
+45 -9
View File
@@ -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<EntityRelationSpec>();
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));
plugins/catalog-backend/src/ingestion/processors/types.ts
+1
View File
@@ -120,6 +120,7 @@ export type CatalogProcessorEntityResult = {
export type CatalogProcessorRelationResult = {
type: 'relation';
relation: EntityRelationSpec;
entityRef?: string;
};
export type CatalogProcessorErrorResult = {
plugins/catalog-backend/src/ingestion/types.ts
+1 -1
View File
@@ -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 = {
plugins/catalog-backend/src/service/router.test.ts
-3
View File
@@ -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 = {
plugins/catalog-backend/src/service/router.ts
+7 -2
View File
@@ -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;