From d94b7c3dce9c0f243bca596265ce818b58781a10 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Mon, 20 Sep 2021 13:49:53 +0200 Subject: [PATCH 01/15] catalog-backend: inital implementation of caching in UrlReaderProcessor + types MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Fredrik Adelöw Co-authored-by: blam Co-authored-by: Johan Haals Signed-off-by: Patrik Oldsberg --- .../processors/UrlReaderProcessor.ts | 50 +++++++++++--- .../src/ingestion/processors/types.ts | 65 ++++++++++++++----- 2 files changed, 92 insertions(+), 23 deletions(-) diff --git a/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.ts b/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.ts index e3ec8c0480..34ddb544ea 100644 --- a/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.ts +++ b/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.ts @@ -22,15 +22,25 @@ import { Logger } from 'winston'; import * as result from './results'; import { CatalogProcessor, + CatalogProcessorCache, CatalogProcessorEmit, + CatalogProcessorEntityResult, CatalogProcessorParser, + CatalogProcessorResult, } from './types'; +const CACHE_KEY = 'v1'; + type Options = { reader: UrlReader; logger: Logger; }; +type CacheItem = { + etag: string; + value: CatalogProcessorEntityResult[]; +}; + export class UrlReaderProcessor implements CatalogProcessor { constructor(private readonly options: Options) {} @@ -39,25 +49,48 @@ export class UrlReaderProcessor implements CatalogProcessor { optional: boolean, emit: CatalogProcessorEmit, parser: CatalogProcessorParser, + cache: CatalogProcessorCache, ): Promise { if (location.type !== 'url') { return false; } + const cacheItem = await cache.get(CACHE_KEY); + try { - const output = await this.doRead(location.target); + const [output, newEtag] = await this.doRead( + location.target, + cacheItem?.etag, + ); + + const parseResults: CatalogProcessorResult[] = []; for (const item of output) { for await (const parseResult of parser({ data: item.data, location: { type: location.type, target: item.url }, })) { + parseResults.push(parseResult); emit(parseResult); } } + + const isOnlyEntities = parseResults.every( + (r): r is CatalogProcessorEntityResult => r.type === 'entity', + ); + if (newEtag && isOnlyEntities) { + await cache.set(CACHE_KEY, { + etag: newEtag, + value: parseResults, + }); + } } catch (error) { const message = `Unable to read ${location.type}, ${error}`; - - if (error.name === 'NotFoundError') { + if (error.name === 'NotModifiedError' && cacheItem) { + for (const parseResult of cacheItem.value) { + emit(parseResult); + } + cache.set(CACHE_KEY, cacheItem); + } else if (error.name === 'NotFoundError') { if (!optional) { emit(result.notFoundError(location, message)); } @@ -71,27 +104,28 @@ export class UrlReaderProcessor implements CatalogProcessor { private async doRead( location: string, - ): Promise<{ data: Buffer; url: string }[]> { + etag?: string, + ): Promise<[response: { data: Buffer; url: string }[], etag?: string]> { // Does it contain globs? I.e. does it contain asterisks or question marks // (no curly braces for now) const { filepath } = parseGitUrl(location); if (filepath?.match(/[*?]/)) { const limiter = limiterFactory(5); - const response = await this.options.reader.search(location); + const response = await this.options.reader.search(location, { etag }); const output = response.files.map(async file => ({ url: file.url, data: await limiter(file.content), })); - return Promise.all(output); + return [await Promise.all(output), response.etag]; } // Otherwise do a plain read, prioritizing readUrl if available if (this.options.reader.readUrl) { const data = await this.options.reader.readUrl(location); - return [{ url: location, data: await data.buffer() }]; + return [[{ url: location, data: await data.buffer() }], data.etag]; } const data = await this.options.reader.read(location); - return [{ url: location, data }]; + return [[{ url: location, data }]]; } } diff --git a/plugins/catalog-backend/src/ingestion/processors/types.ts b/plugins/catalog-backend/src/ingestion/processors/types.ts index 2e2418c8f1..6e1488a36e 100644 --- a/plugins/catalog-backend/src/ingestion/processors/types.ts +++ b/plugins/catalog-backend/src/ingestion/processors/types.ts @@ -19,16 +19,18 @@ import { EntityRelationSpec, LocationSpec, } from '@backstage/catalog-model'; +import { JsonValue } from '@backstage/config'; export type CatalogProcessor = { /** * Reads the contents of a location. * - * @param location The location to read - * @param optional Whether a missing target should trigger an error - * @param emit A sink for items resulting from the read - * @param parser A parser, that is able to take the raw catalog descriptor + * @param location - The location to read + * @param optional - Whether a missing target should trigger an error + * @param emit - A sink for items resulting from the read + * @param parser - A parser, that is able to take the raw catalog descriptor * data and turn it into the actual result pieces. + * @param cache - A cache for storing values local to this processor and the current entity. * @returns True if handled by this processor, false otherwise */ readLocation?( @@ -36,6 +38,7 @@ export type CatalogProcessor = { optional: boolean, emit: CatalogProcessorEmit, parser: CatalogProcessorParser, + cache: CatalogProcessorCache, ): Promise; /** @@ -46,12 +49,13 @@ export type CatalogProcessor = { * additional data, and the input entity may actually still be incomplete * when the processor is invoked. * - * @param entity The (possibly partial) entity to process - * @param location The location that the entity came from - * @param emit A sink for auxiliary items resulting from the processing - * @param originLocation The location that the entity originally came from. + * @param entity - The (possibly partial) entity to process + * @param location - The location that the entity came from + * @param emit - A sink for auxiliary items resulting from the processing + * @param originLocation - The location that the entity originally came from. * While location resolves to the direct parent location, originLocation * tells which location was used to start the ingestion loop. + * @param cache - A cache for storing values local to this processor and the current entity. * @returns The same entity or a modified version of it */ preProcessEntity?( @@ -59,13 +63,14 @@ export type CatalogProcessor = { location: LocationSpec, emit: CatalogProcessorEmit, originLocation: LocationSpec, + cache: CatalogProcessorCache, ): Promise; /** * Validates the entity as a known entity kind, after it has been pre- * processed and has passed through basic overall validation. * - * @param entity The entity to validate + * @param entity - The entity to validate * @returns Resolves to true, if the entity was of a kind that was known and * handled by this processor, and was found to be valid. Resolves to false, * if the entity was not of a kind that was known by this processor. @@ -77,23 +82,25 @@ export type CatalogProcessor = { /** * Post-processes an emitted entity, after it has been validated. * - * @param entity The entity to process - * @param location The location that the entity came from - * @param emit A sink for auxiliary items resulting from the processing + * @param entity - The entity to process + * @param location - The location that the entity came from + * @param emit - A sink for auxiliary items resulting from the processing + * @param cache - A cache for storing values local to this processor and the current entity. * @returns The same entity or a modified version of it */ postProcessEntity?( entity: Entity, location: LocationSpec, emit: CatalogProcessorEmit, + cache: CatalogProcessorCache, ): Promise; /** * Handles an emitted error. * - * @param error The error - * @param location The location where the error occurred - * @param emit A sink for items resulting from this handling + * @param error - The error + * @param location - The location where the error occurred + * @param emit - A sink for items resulting from this handling * @returns Nothing */ handleError?( @@ -113,6 +120,34 @@ export type CatalogProcessorParser = (options: { location: LocationSpec; }) => AsyncIterable; +/** + * A cache for storing data during processing. + * + * The values stored in the cache are always local to each processor, meaning + * no processor can see cache values from other processors. + * + * The cache instance provided to the CatalogProcessor is also scoped to the + * entity being processed, meaning that each processor run can't see cache + * values from processing runs for other entities. + * + * Values that are set during a processing run will only be visible in the directly + * following run. The cache will be overwritten every run, meaning existing values + * are removed and need to be set again for them to remain in the cache. + * + * @public + */ +export interface CatalogProcessorCache { + /** + * Retrieve a value from the cache. + */ + get(key: string): Promise; + + /** + * Store a value in the cache. + */ + set(key: string, value: ItemType): Promise; +} + export type CatalogProcessorEmit = (generated: CatalogProcessorResult) => void; export type CatalogProcessorLocationResult = { From ec2cc1ec56854fb4067a6a873662df01f50719e1 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Mon, 20 Sep 2021 14:12:05 +0200 Subject: [PATCH 02/15] catalog-backend: add tests for UrlReaderProcessor caching MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Johan Haals Co-authored-by: blam Co-authored-by: Fredrik Adelöw Signed-off-by: Patrik Oldsberg --- .../processors/UrlReaderProcessor.test.ts | 70 ++++++++++++++++++- 1 file changed, 67 insertions(+), 3 deletions(-) diff --git a/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.test.ts b/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.test.ts index 6521a2b6bf..5874806ac3 100644 --- a/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.test.ts +++ b/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.test.ts @@ -24,6 +24,7 @@ import { msw } from '@backstage/test-utils'; import { rest } from 'msw'; import { setupServer } from 'msw/node'; import { + CatalogProcessorCache, CatalogProcessorEntityResult, CatalogProcessorErrorResult, CatalogProcessorResult, @@ -33,10 +34,17 @@ import { defaultEntityDataParser } from './util/parse'; describe('UrlReaderProcessor', () => { const mockApiOrigin = 'http://localhost'; - + const mockCache: jest.Mocked = { + get: jest.fn(), + set: jest.fn(), + }; const server = setupServer(); msw.setupDefaultHandlers(server); + beforeEach(() => { + jest.resetAllMocks(); + }); + it('should load from url', async () => { const logger = getVoidLogger(); const reader = UrlReaders.default({ @@ -58,7 +66,13 @@ describe('UrlReaderProcessor', () => { ); const generated = (await new Promise(emit => - processor.readLocation(spec, false, emit, defaultEntityDataParser), + processor.readLocation( + spec, + false, + emit, + defaultEntityDataParser, + mockCache, + ), )) as CatalogProcessorEntityResult; expect(generated.type).toBe('entity'); @@ -66,6 +80,49 @@ describe('UrlReaderProcessor', () => { expect(generated.entity).toEqual({ mock: 'entity' }); }); + it('should use cached data when available', async () => { + const logger = getVoidLogger(); + const reader = UrlReaders.default({ + logger, + config: new ConfigReader({ + backend: { reading: { allow: [{ host: 'localhost' }] } }, + }), + }); + server.use( + rest.get(`${mockApiOrigin}/component.yaml`, (_, res, ctx) => + res(ctx.status(304)), + ), + ); + const spec = { + type: 'url', + target: `${mockApiOrigin}/component.yaml`, + }; + const cacheItem = { + etag: 'my-etag', + value: [{ type: 'entity', location: spec, entity: { mock: 'entity' } }], + }; + mockCache.get.mockResolvedValue(cacheItem); + const processor = new UrlReaderProcessor({ reader, logger }); + + const generated = (await new Promise(emit => + processor.readLocation( + spec, + false, + emit, + defaultEntityDataParser, + mockCache, + ), + )) as CatalogProcessorEntityResult; + + expect(generated.type).toBe('entity'); + expect(generated.location).toEqual(spec); + expect(generated.entity).toEqual({ mock: 'entity' }); + expect(mockCache.get).toBeCalledWith('v1'); + expect(mockCache.get).toBeCalledTimes(1); + expect(mockCache.set).toBeCalledWith('v1', cacheItem); + expect(mockCache.set).toBeCalledTimes(1); + }); + it('should fail load from url with error', async () => { const logger = getVoidLogger(); const reader = UrlReaders.default({ @@ -87,7 +144,13 @@ describe('UrlReaderProcessor', () => { ); const generated = (await new Promise(emit => - processor.readLocation(spec, false, emit, defaultEntityDataParser), + processor.readLocation( + spec, + false, + emit, + defaultEntityDataParser, + mockCache, + ), )) as CatalogProcessorErrorResult; expect(generated.type).toBe('error'); @@ -116,6 +179,7 @@ describe('UrlReaderProcessor', () => { false, emit, defaultEntityDataParser, + mockCache, ); expect(reader.search).toBeCalledTimes(1); From 33333c94169e1d349f0568516c74143e7d9ddd9a Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Mon, 20 Sep 2021 16:14:45 +0200 Subject: [PATCH 03/15] catalog-backend: implement caching in the processing orchestrator MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Johan Haals Co-authored-by: Fredrik Adelöw Co-authored-by: blam Signed-off-by: Patrik Oldsberg --- .../src/ingestion/LocationReaders.ts | 10 +++ .../processors/UrlReaderProcessor.ts | 12 +-- .../src/ingestion/processors/types.ts | 6 ++ .../DefaultCatalogProcessingEngine.test.ts | 12 +-- .../next/DefaultCatalogProcessingEngine.ts | 3 +- .../src/next/DefaultLocationService.test.ts | 4 +- .../src/next/DefaultLocationService.ts | 3 +- .../src/next/DefaultRefreshService.test.ts | 2 +- .../DefaultProcessingDatabase.test.ts | 16 ++-- .../database/DefaultProcessingDatabase.ts | 7 +- .../src/next/database/types.ts | 6 +- .../DefaultCatalogProcessingOrchestrator.ts | 78 ++++++++++++++++++- .../src/next/processing/types.ts | 6 +- 13 files changed, 124 insertions(+), 41 deletions(-) diff --git a/plugins/catalog-backend/src/ingestion/LocationReaders.ts b/plugins/catalog-backend/src/ingestion/LocationReaders.ts index 624c7aaa61..b8c9125893 100644 --- a/plugins/catalog-backend/src/ingestion/LocationReaders.ts +++ b/plugins/catalog-backend/src/ingestion/LocationReaders.ts @@ -52,6 +52,13 @@ type Options = { policy: EntityPolicy; }; +const noopCache = { + async get() { + return undefined; + }, + async set() {}, +}; + /** * Implements the reading of a location through a series of processor tasks. */ @@ -167,6 +174,7 @@ export class LocationReaders implements LocationReader { item.optional, validatedEmit, this.options.parser, + noopCache, ) ) { return; @@ -215,6 +223,7 @@ export class LocationReaders implements LocationReader { item.location, emit, originLocation, + noopCache, ); } catch (e) { const message = `Processor ${ @@ -285,6 +294,7 @@ export class LocationReaders implements LocationReader { current, item.location, emit, + noopCache, ); } catch (e) { const message = `Processor ${ diff --git a/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.ts b/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.ts index 34ddb544ea..67bd0715df 100644 --- a/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.ts +++ b/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.ts @@ -44,6 +44,10 @@ type CacheItem = { export class UrlReaderProcessor implements CatalogProcessor { constructor(private readonly options: Options) {} + getProcessorName() { + return 'url-reader'; + } + async readLocation( location: LocationSpec, optional: boolean, @@ -74,13 +78,11 @@ export class UrlReaderProcessor implements CatalogProcessor { } } - const isOnlyEntities = parseResults.every( - (r): r is CatalogProcessorEntityResult => r.type === 'entity', - ); + const isOnlyEntities = parseResults.every(r => r.type === 'entity'); if (newEtag && isOnlyEntities) { await cache.set(CACHE_KEY, { etag: newEtag, - value: parseResults, + value: parseResults as CatalogProcessorEntityResult[], }); } } catch (error) { @@ -121,7 +123,7 @@ export class UrlReaderProcessor implements CatalogProcessor { // Otherwise do a plain read, prioritizing readUrl if available if (this.options.reader.readUrl) { - const data = await this.options.reader.readUrl(location); + const data = await this.options.reader.readUrl(location, { etag }); return [[{ url: location, data: await data.buffer() }], data.etag]; } diff --git a/plugins/catalog-backend/src/ingestion/processors/types.ts b/plugins/catalog-backend/src/ingestion/processors/types.ts index 6e1488a36e..752789703f 100644 --- a/plugins/catalog-backend/src/ingestion/processors/types.ts +++ b/plugins/catalog-backend/src/ingestion/processors/types.ts @@ -22,6 +22,12 @@ import { import { JsonValue } from '@backstage/config'; export type CatalogProcessor = { + /** + * A unique identifier for the Catalog Processor. + * It's strongly recommended implement getProcessorName as this method will be required in the future. + */ + getProcessorName?(): string; + /** * Reads the contents of a location. * diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts index f7dede4adb..b4b309beef 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts @@ -55,7 +55,7 @@ describe('DefaultCatalogProcessingEngine', () => { relations: [], errors: [], deferredEntities: [], - state: new Map(), + state: {}, }); const engine = new DefaultCatalogProcessingEngine( getVoidLogger(), @@ -84,7 +84,7 @@ describe('DefaultCatalogProcessingEngine', () => { metadata: { name: 'test' }, }, resultHash: '', - state: new Map(), + state: {}, nextUpdateAt: DateTime.now(), lastDiscoveryAt: DateTime.now(), }, @@ -117,7 +117,7 @@ describe('DefaultCatalogProcessingEngine', () => { relations: [], errors: [], deferredEntities: [], - state: new Map(), + state: {}, }); const engine = new DefaultCatalogProcessingEngine( getVoidLogger(), @@ -147,7 +147,7 @@ describe('DefaultCatalogProcessingEngine', () => { metadata: { name: 'test' }, }, resultHash: '', - state: new Map(), + state: {}, nextUpdateAt: DateTime.now(), lastDiscoveryAt: DateTime.now(), }, @@ -181,7 +181,7 @@ describe('DefaultCatalogProcessingEngine', () => { entityRef: '', unprocessedEntity: entity, resultHash: 'the matching hash', - state: new Map(), + state: {}, nextUpdateAt: DateTime.now(), lastDiscoveryAt: DateTime.now(), }; @@ -194,7 +194,7 @@ describe('DefaultCatalogProcessingEngine', () => { relations: [], errors: [], deferredEntities: [], - state: new Map(), + state: {}, }); const engine = new DefaultCatalogProcessingEngine( diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index 4d2d74f05b..bbdac66b58 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -167,8 +167,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { hashBuilder = hashBuilder .update(stableStringify({ ...result.completedEntity })) .update(stableStringify([...result.deferredEntities])) - .update(stableStringify([...result.relations])) - .update(stableStringify(Object.fromEntries(result.state))); + .update(stableStringify([...result.relations])); } const resultHash = hashBuilder.digest('hex'); diff --git a/plugins/catalog-backend/src/next/DefaultLocationService.test.ts b/plugins/catalog-backend/src/next/DefaultLocationService.test.ts index 939f54e793..10c9a742c5 100644 --- a/plugins/catalog-backend/src/next/DefaultLocationService.test.ts +++ b/plugins/catalog-backend/src/next/DefaultLocationService.test.ts @@ -39,7 +39,7 @@ describe('DefaultLocationServiceTest', () => { store.listLocations.mockResolvedValueOnce([]); orchestrator.process.mockResolvedValueOnce({ ok: true, - state: new Map(), + state: {}, completedEntity: { apiVersion: 'backstage.io/v1alpha1', kind: 'Location', @@ -65,7 +65,7 @@ describe('DefaultLocationServiceTest', () => { orchestrator.process.mockResolvedValueOnce({ ok: true, - state: new Map(), + state: {}, completedEntity: { apiVersion: 'backstage.io/v1alpha1', kind: 'Component', diff --git a/plugins/catalog-backend/src/next/DefaultLocationService.ts b/plugins/catalog-backend/src/next/DefaultLocationService.ts index 7d485bea40..678ce9791c 100644 --- a/plugins/catalog-backend/src/next/DefaultLocationService.ts +++ b/plugins/catalog-backend/src/next/DefaultLocationService.ts @@ -87,7 +87,6 @@ export class DefaultLocationService implements LocationService { { entity, locationKey: `${spec.type}:${spec.target}` }, ]; const entities: Entity[] = []; - const state = new Map(); // ignored while (unprocessedEntities.length) { const currentEntity = unprocessedEntities.pop(); if (!currentEntity) { @@ -95,7 +94,7 @@ export class DefaultLocationService implements LocationService { } const processed = await this.orchestrator.process({ entity: currentEntity.entity, - state, + state: {}, // we process without the existing cache }); if (processed.ok) { diff --git a/plugins/catalog-backend/src/next/DefaultRefreshService.test.ts b/plugins/catalog-backend/src/next/DefaultRefreshService.test.ts index 5df57c6897..79ced3ea43 100644 --- a/plugins/catalog-backend/src/next/DefaultRefreshService.test.ts +++ b/plugins/catalog-backend/src/next/DefaultRefreshService.test.ts @@ -138,7 +138,7 @@ describe('Refresh integration', () => { relations: [], errors: [], deferredEntities, - state: new Map(), + state: {}, }; }, }, diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts index afc3806d36..a39d2ba981 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts @@ -17,7 +17,6 @@ import { getVoidLogger } from '@backstage/backend-common'; import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils'; import { Entity, stringifyEntityRef } from '@backstage/catalog-model'; -import { JsonObject } from '@backstage/config'; import { Knex } from 'knex'; import * as uuid from 'uuid'; import { Logger } from 'winston'; @@ -213,7 +212,7 @@ describe('Default Processing Database', () => { id, processedEntity, resultHash: '', - state: new Map(), + state: {}, relations: [], deferredEntities: [], }), @@ -232,7 +231,7 @@ describe('Default Processing Database', () => { id, processedEntity, resultHash: '', - state: new Map(), + state: {}, relations: [], deferredEntities: [], locationKey: 'key', @@ -285,8 +284,7 @@ describe('Default Processing Database', () => { last_discovery_at: '2021-04-01 13:37:00', }); - const state = new Map(); - state.set('hello', { t: 'something' }); + const state = { hello: { t: 'something' } }; await db.transaction(tx => db.updateProcessedEntity(tx, { @@ -308,9 +306,7 @@ describe('Default Processing Database', () => { expect(entities[0].processed_entity).toEqual( JSON.stringify(processedEntity), ); - expect(entities[0].cache).toEqual( - JSON.stringify(Object.fromEntries(state)), - ); + expect(entities[0].cache).toEqual(JSON.stringify(state)); expect(entities[0].errors).toEqual("['something broke']"); expect(entities[0].location_key).toEqual('key'); }, @@ -352,7 +348,7 @@ describe('Default Processing Database', () => { id, processedEntity, resultHash: '', - state: new Map(), + state: {}, relations: relations, deferredEntities: [], }), @@ -404,7 +400,7 @@ describe('Default Processing Database', () => { id, processedEntity, resultHash: '', - state: new Map(), + state: {}, relations: [], deferredEntities, }), diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index 8ba9d54b8f..263ba1bdfd 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -15,7 +15,6 @@ */ import { Entity, stringifyEntityRef } from '@backstage/catalog-model'; -import { JsonObject } from '@backstage/config'; import { ConflictError, NotFoundError } from '@backstage/errors'; import { Knex } from 'knex'; import lodash from 'lodash'; @@ -80,7 +79,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { .update({ processed_entity: JSON.stringify(processedEntity), result_hash: resultHash, - cache: JSON.stringify(Object.fromEntries(state || [])), + cache: JSON.stringify(state), errors, location_key: locationKey, }) @@ -508,9 +507,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { resultHash: i.result_hash || '', nextUpdateAt: timestampToDateTime(i.next_update_at), lastDiscoveryAt: timestampToDateTime(i.last_discovery_at), - state: i.cache - ? JSON.parse(i.cache) - : new Map(), + state: i.cache ? JSON.parse(i.cache) : undefined, errors: i.errors, locationKey: i.location_key, } as RefreshStateItem), diff --git a/plugins/catalog-backend/src/next/database/types.ts b/plugins/catalog-backend/src/next/database/types.ts index 3dfe8e5592..ab449c8a7a 100644 --- a/plugins/catalog-backend/src/next/database/types.ts +++ b/plugins/catalog-backend/src/next/database/types.ts @@ -15,7 +15,7 @@ */ import { Entity, EntityRelationSpec } from '@backstage/catalog-model'; -import { JsonObject } from '@backstage/config'; +import { JsonValue } from '@backstage/config'; import { DateTime } from 'luxon'; import { Transaction } from '../../database/types'; import { DeferredEntity } from '../processing/types'; @@ -36,7 +36,7 @@ export type UpdateProcessedEntityOptions = { id: string; processedEntity: Entity; resultHash: string; - state?: Map; + state?: JsonValue; errors?: string; relations: EntityRelationSpec[]; deferredEntities: DeferredEntity[]; @@ -57,7 +57,7 @@ export type RefreshStateItem = { resultHash: string; nextUpdateAt: DateTime; lastDiscoveryAt: DateTime; // remove? - state: Map; + state?: JsonValue; errors?: string; locationKey?: string; }; diff --git a/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts b/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts index 384060125e..e94c34a660 100644 --- a/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts +++ b/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts @@ -24,6 +24,7 @@ import { stringifyLocationReference, } from '@backstage/catalog-model'; import { ConflictError, InputError, NotAllowedError } from '@backstage/errors'; +import { JsonObject, JsonValue } from '@backstage/config'; import { ScmIntegrationRegistry } from '@backstage/integration'; import path from 'path'; import { Logger } from 'winston'; @@ -32,6 +33,7 @@ import { CatalogProcessorParser, } from '../../ingestion/processors'; import * as results from '../../ingestion/processors/results'; +import { CatalogProcessorCache } from '../../ingestion/processors/types'; import { CatalogProcessingOrchestrator, EntityProcessingRequest, @@ -53,8 +55,70 @@ type Context = { location: LocationSpec; originLocation: LocationSpec; collector: ProcessorOutputCollector; + cache: ProcessorCache; }; +function isObject(value: JsonValue | undefined): value is JsonObject { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +class DefaultCatalogProcessorCache implements CatalogProcessorCache { + private newState?: JsonObject; + constructor(private readonly existingState: JsonObject) {} + + async get( + key: string, + ): Promise { + return this.existingState[key] as ItemType | undefined; + } + + async set( + key: string, + value: ItemType, + ): Promise { + if (!this.newState) { + this.newState = {}; + } + + this.newState[key] = value; + } + + collect(): JsonObject | undefined { + return this.newState; + } +} + +class ProcessorCache { + private caches = new Map(); + + constructor(private readonly existingState: JsonObject) {} + forProcessor(processor: CatalogProcessor): CatalogProcessorCache { + // constructor name will be deprecated in the future when we make `getProcessorName` required in the implementation + const name = processor.getProcessorName?.() ?? processor.constructor.name; + const cache = this.caches.get(name); + if (cache) { + return cache; + } + + const existing = this.existingState[name]; + + const newCache = new DefaultCatalogProcessorCache( + isObject(existing) ? existing : {}, + ); + this.caches.set(name, newCache); + return newCache; + } + + collect(): JsonObject { + const result: JsonObject = {}; + for (const [key, value] of this.caches.entries()) { + result[key] = value.collect(); + } + + return result; + } +} + export class DefaultCatalogProcessingOrchestrator implements CatalogProcessingOrchestrator { @@ -72,17 +136,23 @@ export class DefaultCatalogProcessingOrchestrator async process( request: EntityProcessingRequest, ): Promise { - return this.processSingleEntity(request.entity); + return this.processSingleEntity(request.entity, request.state); } private async processSingleEntity( unprocessedEntity: Entity, + state: JsonValue | undefined, ): Promise { const collector = new ProcessorOutputCollector( this.options.logger, unprocessedEntity, ); + // Cache that is scoped to the entity and processor + const cache = new ProcessorCache( + isObject(state) && isObject(state.cache) ? state.cache : {}, + ); + try { // This will be checked and mutated step by step below let entity: Entity = unprocessedEntity; @@ -108,6 +178,7 @@ export class DefaultCatalogProcessingOrchestrator originLocation: parseLocationReference( getEntityOriginLocationRef(entity), ), + cache, collector, }; @@ -145,7 +216,7 @@ export class DefaultCatalogProcessingOrchestrator return { ...collectorResults, completedEntity: entity, - state: new Map(), + state: { cache: cache.collect() }, ok: true, }; } catch (error) { @@ -173,6 +244,7 @@ export class DefaultCatalogProcessingOrchestrator context.location, context.collector.onEmit, context.originLocation, + context.cache.forProcessor(processor), ); } catch (e) { throw new InputError( @@ -306,6 +378,7 @@ export class DefaultCatalogProcessingOrchestrator false, context.collector.onEmit, this.options.parser, + context.cache.forProcessor(processor), ); if (read) { didRead = true; @@ -343,6 +416,7 @@ export class DefaultCatalogProcessingOrchestrator result, context.location, context.collector.onEmit, + context.cache.forProcessor(processor), ); } catch (e) { throw new InputError( diff --git a/plugins/catalog-backend/src/next/processing/types.ts b/plugins/catalog-backend/src/next/processing/types.ts index 9e30d07404..59a10c084f 100644 --- a/plugins/catalog-backend/src/next/processing/types.ts +++ b/plugins/catalog-backend/src/next/processing/types.ts @@ -15,17 +15,17 @@ */ import { Entity, EntityRelationSpec } from '@backstage/catalog-model'; -import { JsonObject } from '@backstage/config'; +import { JsonValue } from '@backstage/config'; export type EntityProcessingRequest = { entity: Entity; - state: Map; // Versions for multiple deployments etc + state?: JsonValue; // Versions for multiple deployments etc }; export type EntityProcessingResult = | { ok: true; - state: Map; + state: JsonValue; completedEntity: Entity; deferredEntities: DeferredEntity[]; relations: EntityRelationSpec[]; From c9c284b0e0a1978a1890a9b5bb1d70db9821dd42 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Tue, 21 Sep 2021 13:09:53 +0200 Subject: [PATCH 04/15] catalog-backend: split out ProcessorCacheManager and add tests Signed-off-by: Patrik Oldsberg --- .../DefaultCatalogProcessingOrchestrator.ts | 70 ++--------------- .../processing/ProcessorCacheManager.test.ts | 62 +++++++++++++++ .../next/processing/ProcessorCacheManager.ts | 78 +++++++++++++++++++ .../src/next/processing/util.ts | 5 ++ 4 files changed, 150 insertions(+), 65 deletions(-) create mode 100644 plugins/catalog-backend/src/next/processing/ProcessorCacheManager.test.ts create mode 100644 plugins/catalog-backend/src/next/processing/ProcessorCacheManager.ts diff --git a/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts b/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts index e94c34a660..01640f34b1 100644 --- a/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts +++ b/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts @@ -24,7 +24,7 @@ import { stringifyLocationReference, } from '@backstage/catalog-model'; import { ConflictError, InputError, NotAllowedError } from '@backstage/errors'; -import { JsonObject, JsonValue } from '@backstage/config'; +import { JsonValue } from '@backstage/config'; import { ScmIntegrationRegistry } from '@backstage/integration'; import path from 'path'; import { Logger } from 'winston'; @@ -33,7 +33,6 @@ import { CatalogProcessorParser, } from '../../ingestion/processors'; import * as results from '../../ingestion/processors/results'; -import { CatalogProcessorCache } from '../../ingestion/processors/types'; import { CatalogProcessingOrchestrator, EntityProcessingRequest, @@ -47,78 +46,19 @@ import { toAbsoluteUrl, validateEntity, validateEntityEnvelope, + isObject, } from './util'; import { CatalogRulesEnforcer } from '../../ingestion/CatalogRules'; +import { ProcessorCacheManager } from './ProcessorCacheManager'; type Context = { entityRef: string; location: LocationSpec; originLocation: LocationSpec; collector: ProcessorOutputCollector; - cache: ProcessorCache; + cache: ProcessorCacheManager; }; -function isObject(value: JsonValue | undefined): value is JsonObject { - return typeof value === 'object' && value !== null && !Array.isArray(value); -} - -class DefaultCatalogProcessorCache implements CatalogProcessorCache { - private newState?: JsonObject; - constructor(private readonly existingState: JsonObject) {} - - async get( - key: string, - ): Promise { - return this.existingState[key] as ItemType | undefined; - } - - async set( - key: string, - value: ItemType, - ): Promise { - if (!this.newState) { - this.newState = {}; - } - - this.newState[key] = value; - } - - collect(): JsonObject | undefined { - return this.newState; - } -} - -class ProcessorCache { - private caches = new Map(); - - constructor(private readonly existingState: JsonObject) {} - forProcessor(processor: CatalogProcessor): CatalogProcessorCache { - // constructor name will be deprecated in the future when we make `getProcessorName` required in the implementation - const name = processor.getProcessorName?.() ?? processor.constructor.name; - const cache = this.caches.get(name); - if (cache) { - return cache; - } - - const existing = this.existingState[name]; - - const newCache = new DefaultCatalogProcessorCache( - isObject(existing) ? existing : {}, - ); - this.caches.set(name, newCache); - return newCache; - } - - collect(): JsonObject { - const result: JsonObject = {}; - for (const [key, value] of this.caches.entries()) { - result[key] = value.collect(); - } - - return result; - } -} - export class DefaultCatalogProcessingOrchestrator implements CatalogProcessingOrchestrator { @@ -149,7 +89,7 @@ export class DefaultCatalogProcessingOrchestrator ); // Cache that is scoped to the entity and processor - const cache = new ProcessorCache( + const cache = new ProcessorCacheManager( isObject(state) && isObject(state.cache) ? state.cache : {}, ); diff --git a/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.test.ts b/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.test.ts new file mode 100644 index 0000000000..bfdfe93d05 --- /dev/null +++ b/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.test.ts @@ -0,0 +1,62 @@ +/* + * Copyright 2021 The Backstage Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { CatalogProcessor } from '../../ingestion/processors'; +import { ProcessorCacheManager } from './ProcessorCacheManager'; + +class MyProcessor implements CatalogProcessor { + getProcessorName = () => 'my-processor'; +} + +class OtherProcessor implements CatalogProcessor {} + +describe('ProcessorCacheManager', () => { + const myProcessor = new MyProcessor(); + const otherProcessor = new OtherProcessor(); + + it('should forward existing state and collect new state', async () => { + const cache = new ProcessorCacheManager({ + 'my-processor': { 'my-key': 'my-value' }, + }); + + // Should be empty to begin with + expect(cache.collect()).toEqual({}); + + // instance should be cached + const processorCache = cache.forProcessor(myProcessor); + expect(processorCache).toBe(cache.forProcessor(myProcessor)); + + // Initial values should be visible, writes should not + await expect(processorCache.get('my-key')).resolves.toBe( + 'my-value', + ); + processorCache.set('my-key', 'my-new-value'); + await expect(processorCache.get('my-key')).resolves.toBe( + 'my-value', + ); + + // There should be isolation between processors + await expect( + cache.forProcessor(otherProcessor).get('my-key'), + ).resolves.toBeUndefined(); + + // Collecting the state and passing it to a new manager should make the new values visible + const newCache = new ProcessorCacheManager(cache.collect()); + await expect( + newCache.forProcessor(myProcessor).get('my-key'), + ).resolves.toBe('my-new-value'); + }); +}); diff --git a/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.ts b/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.ts new file mode 100644 index 0000000000..7eddc102d2 --- /dev/null +++ b/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.ts @@ -0,0 +1,78 @@ +/* + * Copyright 2021 The Backstage Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { JsonObject, JsonValue } from '@backstage/config'; +import { CatalogProcessor } from '../../ingestion/processors'; +import { CatalogProcessorCache } from '../../ingestion/processors/types'; +import { isObject } from './util'; + +class SingleProcessorCache implements CatalogProcessorCache { + private newState?: JsonObject; + constructor(private readonly existingState: JsonObject) {} + + async get( + key: string, + ): Promise { + return this.existingState[key] as ItemType | undefined; + } + + async set( + key: string, + value: ItemType, + ): Promise { + if (!this.newState) { + this.newState = {}; + } + + this.newState[key] = value; + } + + collect(): JsonObject | undefined { + return this.newState; + } +} + +export class ProcessorCacheManager { + private caches = new Map(); + + constructor(private readonly existingState: JsonObject) {} + + forProcessor(processor: CatalogProcessor): CatalogProcessorCache { + // constructor name will be deprecated in the future when we make `getProcessorName` required in the implementation + const name = processor.getProcessorName?.() ?? processor.constructor.name; + const cache = this.caches.get(name); + if (cache) { + return cache; + } + + const existing = this.existingState[name]; + + const newCache = new SingleProcessorCache( + isObject(existing) ? existing : {}, + ); + this.caches.set(name, newCache); + return newCache; + } + + collect(): JsonObject { + const result: JsonObject = {}; + for (const [key, value] of this.caches.entries()) { + result[key] = value.collect(); + } + + return result; + } +} diff --git a/plugins/catalog-backend/src/next/processing/util.ts b/plugins/catalog-backend/src/next/processing/util.ts index 8b06a76d16..ad8fe8be60 100644 --- a/plugins/catalog-backend/src/next/processing/util.ts +++ b/plugins/catalog-backend/src/next/processing/util.ts @@ -24,6 +24,7 @@ import { ORIGIN_LOCATION_ANNOTATION, stringifyEntityRef, } from '@backstage/catalog-model'; +import { JsonObject, JsonValue } from '@backstage/config'; import { InputError } from '@backstage/errors'; import { ScmIntegrationRegistry } from '@backstage/integration'; import path from 'path'; @@ -76,6 +77,10 @@ export function toAbsoluteUrl( } } +export function isObject(value: JsonValue | undefined): value is JsonObject { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + export const validateEntity = entitySchemaValidator(); export const validateEntityEnvelope = entityEnvelopeSchemaValidator(); From 1572d02b63ea458f1bbed4b588368b02512d02f5 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Tue, 21 Sep 2021 13:24:17 +0200 Subject: [PATCH 05/15] catalog-backend: add changeset for caching + update API report Signed-off-by: Patrik Oldsberg --- .changeset/seven-bulldogs-grin.md | 11 +++++++++ plugins/catalog-backend/api-report.md | 33 ++++++++++++--------------- 2 files changed, 26 insertions(+), 18 deletions(-) create mode 100644 .changeset/seven-bulldogs-grin.md diff --git a/.changeset/seven-bulldogs-grin.md b/.changeset/seven-bulldogs-grin.md new file mode 100644 index 0000000000..2d6f0c14c1 --- /dev/null +++ b/.changeset/seven-bulldogs-grin.md @@ -0,0 +1,11 @@ +--- +'@backstage/plugin-catalog-backend': minor +--- + +Introduced a new `CatalogProcessorCache` that is available to catalog processors. It allows arbitrary values to be saved that will then be visible during the next run. The cache is scoped to each individual processor and entity, but is shared across processing steps in a single processor. + +The cache is available as a new argument to each of the processing steps, except for `validateEntityKind` and `handleError`. + +This also introduces an optional `getProcessorName` to the `CatalogProcessor` interface, which is used to provide a stable identifier for the processor. While it is currently optional it will move to be required in the future. + +The breaking part of this change is the modification of the `state` field in the `EntityProcessingRequest` and `EntityProcessingResult` types. This is unlikely to have any impact as the `state` field was previously unused, but could require some minor updates. diff --git a/plugins/catalog-backend/api-report.md b/plugins/catalog-backend/api-report.md index 2422221d6d..10f149f850 100644 --- a/plugins/catalog-backend/api-report.md +++ b/plugins/catalog-backend/api-report.md @@ -17,7 +17,6 @@ import { EntityPolicy } from '@backstage/catalog-model'; import { EntityRelationSpec } from '@backstage/catalog-model'; import express from 'express'; import { IndexableDocument } from '@backstage/search-common'; -import { JsonObject } from '@backstage/config'; import { JsonValue } from '@backstage/config'; import { Knex } from 'knex'; import { Location as Location_2 } from '@backstage/catalog-model'; @@ -302,23 +301,27 @@ export interface CatalogProcessingOrchestrator { // // @public (undocumented) export type CatalogProcessor = { + getProcessorName?(): string; readLocation?( location: LocationSpec, optional: boolean, emit: CatalogProcessorEmit, parser: CatalogProcessorParser, + cache: CatalogProcessorCache, ): Promise; preProcessEntity?( entity: Entity, location: LocationSpec, emit: CatalogProcessorEmit, originLocation: LocationSpec, + cache: CatalogProcessorCache, ): Promise; validateEntityKind?(entity: Entity): Promise; postProcessEntity?( entity: Entity, location: LocationSpec, emit: CatalogProcessorEmit, + cache: CatalogProcessorCache, ): Promise; handleError?( error: Error, @@ -327,6 +330,12 @@ export type CatalogProcessor = { ): Promise; }; +// @public +export interface CatalogProcessorCache { + get(key: string): Promise; + set(key: string, value: ItemType): Promise; +} + // Warning: (ae-missing-release-tag) "CatalogProcessorEmit" is exported by the package, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -868,7 +877,7 @@ export type EntityPagination = { // @public (undocumented) export type EntityProcessingRequest = { entity: Entity; - state: Map; + state?: JsonValue; }; // Warning: (ae-missing-release-tag) "EntityProcessingResult" is exported by the package, but it is missing a release tag (@alpha, @beta, @public, or @internal) @@ -877,7 +886,7 @@ export type EntityProcessingRequest = { export type EntityProcessingResult = | { ok: true; - state: Map; + state: JsonValue; completedEntity: Entity; deferredEntities: DeferredEntity[]; relations: EntityRelationSpec[]; @@ -1478,11 +1487,14 @@ export class UrlReaderProcessor implements CatalogProcessor { // Warning: (ae-forgotten-export) The symbol "Options" needs to be exported by the entry point index.d.ts constructor(options: Options_3); // (undocumented) + getProcessorName(): string; + // (undocumented) readLocation( location: LocationSpec, optional: boolean, emit: CatalogProcessorEmit, parser: CatalogProcessorParser, + cache: CatalogProcessorCache, ): Promise; } @@ -1505,21 +1517,6 @@ export class UrlReaderProcessor implements CatalogProcessor { // src/database/types.d.ts:164:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen // src/database/types.d.ts:165:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen // src/ingestion/processors/GithubMultiOrgReaderProcessor.d.ts:23:9 - (ae-forgotten-export) The symbol "GithubMultiOrgConfig" needs to be exported by the entry point index.d.ts -// src/ingestion/processors/types.d.ts:7:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen -// src/ingestion/processors/types.d.ts:8:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen -// src/ingestion/processors/types.d.ts:9:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen -// src/ingestion/processors/types.d.ts:10:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen -// src/ingestion/processors/types.d.ts:23:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen -// src/ingestion/processors/types.d.ts:24:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen -// src/ingestion/processors/types.d.ts:25:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen -// src/ingestion/processors/types.d.ts:26:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen -// src/ingestion/processors/types.d.ts:36:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen -// src/ingestion/processors/types.d.ts:47:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen -// src/ingestion/processors/types.d.ts:48:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen -// src/ingestion/processors/types.d.ts:49:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen -// src/ingestion/processors/types.d.ts:56:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen -// src/ingestion/processors/types.d.ts:57:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen -// src/ingestion/processors/types.d.ts:58:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen // src/ingestion/types.d.ts:17:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen // src/ingestion/types.d.ts:41:8 - (tsdoc-param-tag-missing-hyphen) The @param block should be followed by a parameter name and then a hyphen ``` From f8a515a790a2f57f8881414c853cbfbb513b930d Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Tue, 21 Sep 2021 13:54:59 +0200 Subject: [PATCH 06/15] catalog-backend: added some basic tests for the default orchestrator Signed-off-by: Patrik Oldsberg --- .../DefaultCatalogProcessingEngine.test.ts | 8 +- ...faultCatalogProcessingOrchestrator.test.ts | 186 ++++++++++++++++-- 2 files changed, 178 insertions(+), 16 deletions(-) diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts index b4b309beef..c569bf8135 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts @@ -84,7 +84,7 @@ describe('DefaultCatalogProcessingEngine', () => { metadata: { name: 'test' }, }, resultHash: '', - state: {}, + state: [], nextUpdateAt: DateTime.now(), lastDiscoveryAt: DateTime.now(), }, @@ -100,7 +100,7 @@ describe('DefaultCatalogProcessingEngine', () => { kind: 'Location', metadata: { name: 'test' }, }, - state: expect.anything(), + state: [], // State is forwarded as is, even if it's a bad format }); }); await engine.stop(); @@ -147,7 +147,7 @@ describe('DefaultCatalogProcessingEngine', () => { metadata: { name: 'test' }, }, resultHash: '', - state: {}, + state: { cache: { myProcessor: { myKey: 'myValue' } } }, nextUpdateAt: DateTime.now(), lastDiscoveryAt: DateTime.now(), }, @@ -163,7 +163,7 @@ describe('DefaultCatalogProcessingEngine', () => { kind: 'Location', metadata: { name: 'test' }, }, - state: expect.anything(), + state: { cache: { myProcessor: { myKey: 'myValue' } } }, }); }); await engine.stop(); diff --git a/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.test.ts b/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.test.ts index c93734b37f..f8cf65d4e8 100644 --- a/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.test.ts +++ b/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.test.ts @@ -16,21 +16,186 @@ import { getVoidLogger } from '@backstage/backend-common'; import { + Entity, + EntityPolicies, EntityPolicy, LocationEntity, + LocationSpec, LOCATION_ANNOTATION, ORIGIN_LOCATION_ANNOTATION, } from '@backstage/catalog-model'; -import { ScmIntegrationRegistry } from '@backstage/integration'; +import { + ScmIntegrationRegistry, + ScmIntegrations, +} from '@backstage/integration'; import { CatalogProcessor, + CatalogProcessorCache, + CatalogProcessorEmit, CatalogProcessorParser, results, } from '../../ingestion'; import { CatalogRulesEnforcer } from '../../ingestion/CatalogRules'; import { DefaultCatalogProcessingOrchestrator } from './DefaultCatalogProcessingOrchestrator'; +import { defaultEntityDataParser } from '../../ingestion/processors/util/parse'; +import { ConfigReader } from '@backstage/config'; + +class FooBarProcessor implements CatalogProcessor { + getProcessorName = () => 'foo-bar'; + + async validateEntityKind(entity: Entity) { + return entity.kind.toLocaleLowerCase('en-US') === 'foobar'; + } + + async postProcessEntity( + entity: Entity, + _location: LocationSpec, + emit: CatalogProcessorEmit, + cache: CatalogProcessorCache, + ) { + if (await cache.get('emit')) { + emit( + results.entity( + { type: 'url', target: './new-place' }, + { + apiVersion: 'my-api/v1', + kind: 'FooBar', + metadata: { + name: 'my-new-foo-bar', + }, + }, + ), + ); + emit( + results.relation({ + type: 'my-type', + source: { kind: 'foobar', name: 'my-source', namespace: 'default' }, + target: { kind: 'foobar', name: 'my-target', namespace: 'default' }, + }), + ); + } + return entity; + } +} describe('DefaultCatalogProcessingOrchestrator', () => { + describe('2', () => { + const entity = { + apiVersion: 'my-api/v1', + kind: 'FooBar', + metadata: { + name: 'my-foo-bar', + annotations: { + [LOCATION_ANNOTATION]: 'url:./here', + [ORIGIN_LOCATION_ANNOTATION]: 'url:./there', + }, + }, + }; + + const rulesEnforcer: CatalogRulesEnforcer = { + isAllowed: () => true, + }; + + const orchestrator = new DefaultCatalogProcessingOrchestrator({ + processors: [new FooBarProcessor()], + integrations: ScmIntegrations.fromConfig(new ConfigReader({})), + logger: getVoidLogger(), + parser: defaultEntityDataParser, + policy: EntityPolicies.allOf([]), + rulesEnforcer, + }); + + it('runs a minimal processing', async () => { + await expect(orchestrator.process({ entity })).resolves.toEqual({ + ok: true, + completedEntity: entity, + deferredEntities: [], + errors: [], + relations: [], + state: { + cache: {}, + }, + }); + }); + + it('emits some things', async () => { + await expect( + orchestrator.process({ + entity, + state: { cache: { 'foo-bar': { emit: true } } }, + }), + ).resolves.toEqual({ + ok: true, + completedEntity: entity, + deferredEntities: [ + { + locationKey: 'url:./new-place', + entity: { + apiVersion: 'my-api/v1', + kind: 'FooBar', + metadata: { + name: 'my-new-foo-bar', + annotations: { + [LOCATION_ANNOTATION]: 'url:./new-place', + [ORIGIN_LOCATION_ANNOTATION]: 'url:./there', + }, + }, + }, + }, + ], + errors: [], + relations: [ + { + type: 'my-type', + source: { kind: 'foobar', name: 'my-source', namespace: 'default' }, + target: { kind: 'foobar', name: 'my-target', namespace: 'default' }, + }, + ], + state: { + cache: { 'foo-bar': { emit: true } }, + }, + }); + }); + + it('accepts any state input', async () => { + await expect( + orchestrator.process({ entity, state: null as any }), + ).resolves.toMatchObject({ + ok: true, + }); + await expect( + orchestrator.process({ entity, state: [] as any }), + ).resolves.toMatchObject({ + ok: true, + }); + await expect( + orchestrator.process({ entity, state: Symbol() as any }), + ).resolves.toMatchObject({ + ok: true, + }); + await expect( + orchestrator.process({ entity, state: undefined }), + ).resolves.toMatchObject({ + ok: true, + }); + await expect( + orchestrator.process({ entity, state: 3 as any }), + ).resolves.toMatchObject({ + ok: true, + }); + await expect( + orchestrator.process({ entity, state: '}{' as any }), + ).resolves.toMatchObject({ + ok: true, + }); + await expect( + orchestrator.process({ entity, state: { cache: null } }), + ).resolves.toMatchObject({ + ok: true, + }); + }); + }); + it('enforces catalog rules', async () => { const entity: LocationEntity = { apiVersion: 'backstage.io/v1beta1', @@ -48,6 +213,7 @@ describe('DefaultCatalogProcessingOrchestrator', () => { }, }; + const integrations = ScmIntegrations.fromConfig(new ConfigReader({})); const processor: jest.Mocked = { validateEntityKind: jest.fn(async () => true), readLocation: jest.fn(async (_l, _o, emit) => { @@ -55,11 +221,7 @@ describe('DefaultCatalogProcessingOrchestrator', () => { return true; }), }; - const integrations: jest.Mocked = {} as any; const parser: CatalogProcessorParser = jest.fn(); - const policy: jest.Mocked = { - enforce: jest.fn(async x => x), - }; const rulesEnforcer: jest.Mocked = { isAllowed: jest.fn(), }; @@ -69,18 +231,18 @@ describe('DefaultCatalogProcessingOrchestrator', () => { integrations, logger: getVoidLogger(), parser, - policy, + policy: EntityPolicies.allOf([]), rulesEnforcer, }); rulesEnforcer.isAllowed.mockReturnValueOnce(true); - await expect( - orchestrator.process({ entity, state: new Map() }), - ).resolves.toEqual(expect.objectContaining({ ok: true })); + await expect(orchestrator.process({ entity, state: {} })).resolves.toEqual( + expect.objectContaining({ ok: true }), + ); rulesEnforcer.isAllowed.mockReturnValueOnce(false); - await expect( - orchestrator.process({ entity, state: new Map() }), - ).resolves.toEqual(expect.objectContaining({ ok: false })); + await expect(orchestrator.process({ entity, state: {} })).resolves.toEqual( + expect.objectContaining({ ok: false }), + ); }); }); From f0649d43e50b41697a82d9a95448efb6309cc932 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Tue, 21 Sep 2021 17:25:37 +0200 Subject: [PATCH 07/15] catalog-backend: review fixes Signed-off-by: Patrik Oldsberg --- .../ingestion/processors/UrlReaderProcessor.ts | 17 ++++++++++------- .../src/ingestion/processors/types.ts | 2 +- 2 files changed, 11 insertions(+), 8 deletions(-) diff --git a/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.ts b/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.ts index 67bd0715df..a555c54ecd 100644 --- a/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.ts +++ b/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.ts @@ -62,13 +62,13 @@ export class UrlReaderProcessor implements CatalogProcessor { const cacheItem = await cache.get(CACHE_KEY); try { - const [output, newEtag] = await this.doRead( + const { response, etag: newEtag } = await this.doRead( location.target, cacheItem?.etag, ); const parseResults: CatalogProcessorResult[] = []; - for (const item of output) { + for (const item of response) { for await (const parseResult of parser({ data: item.data, location: { type: location.type, target: item.url }, @@ -91,7 +91,7 @@ export class UrlReaderProcessor implements CatalogProcessor { for (const parseResult of cacheItem.value) { emit(parseResult); } - cache.set(CACHE_KEY, cacheItem); + await cache.set(CACHE_KEY, cacheItem); } else if (error.name === 'NotFoundError') { if (!optional) { emit(result.notFoundError(location, message)); @@ -107,7 +107,7 @@ export class UrlReaderProcessor implements CatalogProcessor { private async doRead( location: string, etag?: string, - ): Promise<[response: { data: Buffer; url: string }[], etag?: string]> { + ): Promise<{ response: { data: Buffer; url: string }[]; etag?: string }> { // Does it contain globs? I.e. does it contain asterisks or question marks // (no curly braces for now) const { filepath } = parseGitUrl(location); @@ -118,16 +118,19 @@ export class UrlReaderProcessor implements CatalogProcessor { url: file.url, data: await limiter(file.content), })); - return [await Promise.all(output), response.etag]; + return { response: await Promise.all(output), etag: response.etag }; } // Otherwise do a plain read, prioritizing readUrl if available if (this.options.reader.readUrl) { const data = await this.options.reader.readUrl(location, { etag }); - return [[{ url: location, data: await data.buffer() }], data.etag]; + return { + response: [{ url: location, data: await data.buffer() }], + etag: data.etag, + }; } const data = await this.options.reader.read(location); - return [[{ url: location, data }]]; + return { response: [{ url: location, data }] }; } } diff --git a/plugins/catalog-backend/src/ingestion/processors/types.ts b/plugins/catalog-backend/src/ingestion/processors/types.ts index 752789703f..7c201ca319 100644 --- a/plugins/catalog-backend/src/ingestion/processors/types.ts +++ b/plugins/catalog-backend/src/ingestion/processors/types.ts @@ -24,7 +24,7 @@ import { JsonValue } from '@backstage/config'; export type CatalogProcessor = { /** * A unique identifier for the Catalog Processor. - * It's strongly recommended implement getProcessorName as this method will be required in the future. + * It's strongly recommended to implement getProcessorName as this method will be required in the future. */ getProcessorName?(): string; From c65552b3b7db043536d1268302530af0de7e0f6d Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Tue, 21 Sep 2021 17:26:04 +0200 Subject: [PATCH 08/15] catalog-backend: include cache state in processing result hash Signed-off-by: Patrik Oldsberg --- .../catalog-backend/src/next/DefaultCatalogProcessingEngine.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index bbdac66b58..1955220b2b 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -167,7 +167,8 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { hashBuilder = hashBuilder .update(stableStringify({ ...result.completedEntity })) .update(stableStringify([...result.deferredEntities])) - .update(stableStringify([...result.relations])); + .update(stableStringify([...result.relations])) + .update(stableStringify(result.state)); } const resultHash = hashBuilder.digest('hex'); From 029d8130c1852de740bcf2999f1d2615768477f6 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Tue, 21 Sep 2021 19:28:06 +0200 Subject: [PATCH 09/15] catalog-backend: fix merge conflict Signed-off-by: Patrik Oldsberg --- .../catalog-backend/src/next/DefaultLocationService.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/plugins/catalog-backend/src/next/DefaultLocationService.test.ts b/plugins/catalog-backend/src/next/DefaultLocationService.test.ts index 10c9a742c5..78dd9c1009 100644 --- a/plugins/catalog-backend/src/next/DefaultLocationService.test.ts +++ b/plugins/catalog-backend/src/next/DefaultLocationService.test.ts @@ -124,7 +124,7 @@ describe('DefaultLocationServiceTest', () => { }; orchestrator.process.mockResolvedValueOnce({ ok: true, - state: new Map(), + state: {}, completedEntity: { apiVersion: 'backstage.io/v1alpha1', kind: 'Component', @@ -151,7 +151,7 @@ describe('DefaultLocationServiceTest', () => { it('should return exists false when the location does not exist beforehand', async () => { orchestrator.process.mockResolvedValueOnce({ ok: true, - state: new Map(), + state: {}, completedEntity: { apiVersion: 'backstage.io/v1alpha1', kind: 'Component', From 2be9a2c1676b8ff38eaa86726665dc069d5e22ae Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Wed, 22 Sep 2021 11:41:38 +0200 Subject: [PATCH 10/15] catalog-backend: switch cache to persist items unless new ones are written Co-authored-by: blam Co-authored-by: Johan Haals Signed-off-by: Patrik Oldsberg --- .../processors/UrlReaderProcessor.test.ts | 36 +++++++++++-------- .../processors/UrlReaderProcessor.ts | 1 - .../src/ingestion/processors/types.ts | 4 +-- .../processing/ProcessorCacheManager.test.ts | 19 ++++++++-- .../next/processing/ProcessorCacheManager.ts | 9 ++--- 5 files changed, 45 insertions(+), 24 deletions(-) diff --git a/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.test.ts b/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.test.ts index 5874806ac3..a63a001082 100644 --- a/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.test.ts +++ b/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.test.ts @@ -61,23 +61,30 @@ describe('UrlReaderProcessor', () => { server.use( rest.get(`${mockApiOrigin}/component.yaml`, (_, res, ctx) => - res(ctx.json({ mock: 'entity' })), + res(ctx.set({ ETag: 'my-etag' }), ctx.json({ mock: 'entity' })), ), ); - const generated = (await new Promise(emit => - processor.readLocation( - spec, - false, - emit, - defaultEntityDataParser, - mockCache, - ), - )) as CatalogProcessorEntityResult; + const emitted = new Array(); + await processor.readLocation( + spec, + false, + result => emitted.push(result), + defaultEntityDataParser, + mockCache, + ); - expect(generated.type).toBe('entity'); - expect(generated.location).toEqual(spec); - expect(generated.entity).toEqual({ mock: 'entity' }); + expect(emitted.length).toBe(1); + expect(emitted[0]).toEqual({ + type: 'entity', + location: spec, + entity: { mock: 'entity' }, + }); + expect(mockCache.set).toBeCalledWith('v1', { + etag: 'my-etag', + value: [{ type: 'entity', location: spec, entity: { mock: 'entity' } }], + }); + expect(mockCache.set).toBeCalledTimes(1); }); it('should use cached data when available', async () => { @@ -119,8 +126,7 @@ describe('UrlReaderProcessor', () => { expect(generated.entity).toEqual({ mock: 'entity' }); expect(mockCache.get).toBeCalledWith('v1'); expect(mockCache.get).toBeCalledTimes(1); - expect(mockCache.set).toBeCalledWith('v1', cacheItem); - expect(mockCache.set).toBeCalledTimes(1); + expect(mockCache.set).toBeCalledTimes(0); }); it('should fail load from url with error', async () => { diff --git a/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.ts b/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.ts index a555c54ecd..02aa94522b 100644 --- a/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.ts +++ b/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.ts @@ -91,7 +91,6 @@ export class UrlReaderProcessor implements CatalogProcessor { for (const parseResult of cacheItem.value) { emit(parseResult); } - await cache.set(CACHE_KEY, cacheItem); } else if (error.name === 'NotFoundError') { if (!optional) { emit(result.notFoundError(location, message)); diff --git a/plugins/catalog-backend/src/ingestion/processors/types.ts b/plugins/catalog-backend/src/ingestion/processors/types.ts index 7c201ca319..dccc60d42c 100644 --- a/plugins/catalog-backend/src/ingestion/processors/types.ts +++ b/plugins/catalog-backend/src/ingestion/processors/types.ts @@ -137,8 +137,8 @@ export type CatalogProcessorParser = (options: { * values from processing runs for other entities. * * Values that are set during a processing run will only be visible in the directly - * following run. The cache will be overwritten every run, meaning existing values - * are removed and need to be set again for them to remain in the cache. + * following run. The cache will be overwritten every run unless no new cache items + * are written, in which case the existing values remain in the cache. * * @public */ diff --git a/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.test.ts b/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.test.ts index bfdfe93d05..e7ee27b667 100644 --- a/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.test.ts +++ b/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.test.ts @@ -43,10 +43,25 @@ describe('ProcessorCacheManager', () => { await expect(processorCache.get('my-key')).resolves.toBe( 'my-value', ); - processorCache.set('my-key', 'my-new-value'); + + // If set hasn't been called yet we should get the existing data + expect(cache.collect()).toEqual({ + 'my-processor': { 'my-key': 'my-value' }, + }); + + processorCache.set('my-new-key', 'my-new-value'); + // Once set has been called the old values should disappear + expect(cache.collect()).toEqual({ + 'my-processor': { 'my-new-key': 'my-new-value' }, + }); + + // Getting the cache should return the initial state value await expect(processorCache.get('my-key')).resolves.toBe( 'my-value', ); + await expect( + processorCache.get('my-new-key'), + ).resolves.toBeUndefined(); // There should be isolation between processors await expect( @@ -56,7 +71,7 @@ describe('ProcessorCacheManager', () => { // Collecting the state and passing it to a new manager should make the new values visible const newCache = new ProcessorCacheManager(cache.collect()); await expect( - newCache.forProcessor(myProcessor).get('my-key'), + newCache.forProcessor(myProcessor).get('my-new-key'), ).resolves.toBe('my-new-value'); }); }); diff --git a/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.ts b/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.ts index 7eddc102d2..5da466fe86 100644 --- a/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.ts +++ b/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.ts @@ -21,12 +21,13 @@ import { isObject } from './util'; class SingleProcessorCache implements CatalogProcessorCache { private newState?: JsonObject; - constructor(private readonly existingState: JsonObject) {} + + constructor(private readonly existingState?: JsonObject) {} async get( key: string, ): Promise { - return this.existingState[key] as ItemType | undefined; + return this.existingState?.[key] as ItemType | undefined; } async set( @@ -41,7 +42,7 @@ class SingleProcessorCache implements CatalogProcessorCache { } collect(): JsonObject | undefined { - return this.newState; + return this.newState ?? this.existingState; } } @@ -61,7 +62,7 @@ export class ProcessorCacheManager { const existing = this.existingState[name]; const newCache = new SingleProcessorCache( - isObject(existing) ? existing : {}, + isObject(existing) ? existing : undefined, ); this.caches.set(name, newCache); return newCache; From b562c68091623329029d8a2c3a6ddcaad58f6618 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Wed, 22 Sep 2021 11:50:19 +0200 Subject: [PATCH 11/15] catalog-backend: use inlined local type for cache items Co-authored-by: blam Co-authored-by: Johan Haals Signed-off-by: Patrik Oldsberg --- .../src/ingestion/processors/UrlReaderProcessor.ts | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.ts b/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.ts index 02aa94522b..cf889eb577 100644 --- a/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.ts +++ b/plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.ts @@ -15,7 +15,7 @@ */ import { UrlReader } from '@backstage/backend-common'; -import { LocationSpec } from '@backstage/catalog-model'; +import { Entity, LocationSpec } from '@backstage/catalog-model'; import parseGitUrl from 'git-url-parse'; import limiterFactory from 'p-limit'; import { Logger } from 'winston'; @@ -36,9 +36,14 @@ type Options = { logger: Logger; }; +// WARNING: If you change this type, you likely need to bump the CACHE_KEY as well type CacheItem = { etag: string; - value: CatalogProcessorEntityResult[]; + value: { + type: 'entity'; + entity: Entity; + location: LocationSpec; + }[]; }; export class UrlReaderProcessor implements CatalogProcessor { From 69acb4e7baa4d2de6d76e7139723ba6863d37dca Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Wed, 22 Sep 2021 14:58:55 +0200 Subject: [PATCH 12/15] catalog-backend: separated out state writing + give state a ttl MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Fredrik Adelöw Co-authored-by: blam Co-authored-by: Johan Haals Signed-off-by: Patrik Oldsberg --- .../DefaultCatalogProcessingEngine.test.ts | 89 ++++++++++++++++++- .../next/DefaultCatalogProcessingEngine.ts | 29 +++++- .../DefaultProcessingDatabase.test.ts | 56 ++++++++++-- .../database/DefaultProcessingDatabase.ts | 15 +++- .../src/next/database/types.ts | 18 +++- .../src/next/processing/types.ts | 6 +- 6 files changed, 192 insertions(+), 21 deletions(-) diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts index c569bf8135..430d7f251f 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts @@ -28,6 +28,7 @@ describe('DefaultCatalogProcessingEngine', () => { transaction: jest.fn(), getProcessableEntities: jest.fn(), updateProcessedEntity: jest.fn(), + updateEntityCache: jest.fn(), } as unknown as jest.Mocked; const orchestrator: jest.Mocked = { process: jest.fn(), @@ -84,7 +85,7 @@ describe('DefaultCatalogProcessingEngine', () => { metadata: { name: 'test' }, }, resultHash: '', - state: [], + state: [] as any, nextUpdateAt: DateTime.now(), lastDiscoveryAt: DateTime.now(), }, @@ -221,16 +222,100 @@ describe('DefaultCatalogProcessingEngine', () => { expect(hash.digest).toBeCalledTimes(1); expect(db.updateProcessedEntity).toBeCalledTimes(1); }); + expect(db.updateEntityCache).not.toHaveBeenCalled(); db.getProcessableEntities .mockReset() - .mockResolvedValueOnce({ items: [refreshState] }) + .mockResolvedValueOnce({ + items: [{ ...refreshState, state: { something: 'different' } }], + }) .mockResolvedValue({ items: [] }); await waitForExpect(() => { expect(orchestrator.process).toBeCalledTimes(2); expect(hash.digest).toBeCalledTimes(2); expect(db.updateProcessedEntity).toBeCalledTimes(1); + expect(db.updateEntityCache).toBeCalledTimes(1); + }); + expect(db.updateEntityCache).toHaveBeenCalledWith(expect.anything(), { + id: '', + state: { ttl: 5 }, + }); + await engine.stop(); + }); + + it('should decrease the state ttl if there are errors', async () => { + const entity = { + apiVersion: '1', + kind: 'Location', + metadata: { name: 'test' }, + }; + + const refreshState = { + id: '', + entityRef: '', + unprocessedEntity: entity, + resultHash: 'the matching hash', + state: { some: 'value', ttl: 1 }, + nextUpdateAt: DateTime.now(), + lastDiscoveryAt: DateTime.now(), + }; + + hash.digest.mockReturnValue('the matching hash'); + + orchestrator.process.mockResolvedValue({ + ok: false, + errors: [], + }); + + const engine = new DefaultCatalogProcessingEngine( + getVoidLogger(), + [], + db, + orchestrator, + stitcher, + () => hash, + ); + + db.transaction.mockImplementation(cb => cb((() => {}) as any)); + + await engine.start(); + + db.getProcessableEntities + .mockResolvedValueOnce({ + items: [refreshState], + }) + .mockResolvedValue({ items: [] }); + + await waitForExpect(() => { + expect(db.updateEntityCache).toBeCalledTimes(1); + }); + + expect(db.updateEntityCache).toHaveBeenCalledWith(expect.anything(), { + id: '', + state: { some: 'value', ttl: 0 }, + }); + + // Second run, the TTL should now reach 0 and the cache should be cleared + db.getProcessableEntities + .mockResolvedValueOnce({ + items: [ + { + ...refreshState, + state: db.updateEntityCache.mock.calls[0][1].state, + }, + ], + }) + .mockResolvedValue({ items: [] }); + + db.updateEntityCache.mockReset(); + await waitForExpect(() => { + expect(db.updateEntityCache).toBeCalledTimes(1); + }); + + expect(db.updateEntityCache).toHaveBeenCalledWith(expect.anything(), { + id: '', + state: {}, }); await engine.stop(); diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index 1955220b2b..346a912009 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -38,6 +38,8 @@ import { EntityProviderMutation, } from './types'; +const CACHE_TTL = 5; + class Connection implements EntityProviderConnection { readonly validateEntityEnvelope = entityEnvelopeSchemaValidator(); @@ -151,6 +153,29 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { track.markProcessorsCompleted(result); + if (result.ok) { + if (stableStringify(state) !== stableStringify(result.state)) { + await this.processingDatabase.transaction(async tx => { + await this.processingDatabase.updateEntityCache(tx, { + id, + state: { + ttl: CACHE_TTL, + ...result.state, + }, + }); + }); + } + } else { + const maybeTtl = state?.ttl; + const ttl = Number.isInteger(maybeTtl) ? (maybeTtl as number) : 0; + await this.processingDatabase.transaction(async tx => { + await this.processingDatabase.updateEntityCache(tx, { + id, + state: ttl > 0 ? { ...state, ttl: ttl - 1 } : {}, + }); + }); + } + for (const error of result.errors) { // TODO(freben): Try to extract the location out of the unprocessed // entity and add as meta to the log lines @@ -167,8 +192,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { hashBuilder = hashBuilder .update(stableStringify({ ...result.completedEntity })) .update(stableStringify([...result.deferredEntities])) - .update(stableStringify([...result.relations])) - .update(stableStringify(result.state)); + .update(stableStringify([...result.relations])); } const resultHash = hashBuilder.digest('hex'); @@ -208,7 +232,6 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { id, processedEntity: result.completedEntity, resultHash, - state: result.state, errors: errorsString, relations: result.relations, deferredEntities: result.deferredEntities, diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts index a39d2ba981..c10a37396b 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts @@ -212,7 +212,6 @@ describe('Default Processing Database', () => { id, processedEntity, resultHash: '', - state: {}, relations: [], deferredEntities: [], }), @@ -231,7 +230,6 @@ describe('Default Processing Database', () => { id, processedEntity, resultHash: '', - state: {}, relations: [], deferredEntities: [], locationKey: 'key', @@ -284,14 +282,11 @@ describe('Default Processing Database', () => { last_discovery_at: '2021-04-01 13:37:00', }); - const state = { hello: { t: 'something' } }; - await db.transaction(tx => db.updateProcessedEntity(tx, { id, processedEntity, resultHash: '', - state, relations: [], deferredEntities: [], locationKey: 'key', @@ -306,7 +301,6 @@ describe('Default Processing Database', () => { expect(entities[0].processed_entity).toEqual( JSON.stringify(processedEntity), ); - expect(entities[0].cache).toEqual(JSON.stringify(state)); expect(entities[0].errors).toEqual("['something broke']"); expect(entities[0].location_key).toEqual('key'); }, @@ -348,7 +342,6 @@ describe('Default Processing Database', () => { id, processedEntity, resultHash: '', - state: {}, relations: relations, deferredEntities: [], }), @@ -400,7 +393,6 @@ describe('Default Processing Database', () => { id, processedEntity, resultHash: '', - state: {}, relations: [], deferredEntities, }), @@ -418,6 +410,54 @@ describe('Default Processing Database', () => { ); }); + describe('updateEntityCache', () => { + it.each(databases.eachSupportedId())( + 'updates the entityCache, %p', + async databaseId => { + const { knex, db } = await createDatabase(databaseId); + const id = '123'; + await insertRefreshStateRow(knex, { + entity_id: id, + entity_ref: 'location:default/fakelocation', + unprocessed_entity: '{}', + processed_entity: '{}', + errors: '[]', + next_update_at: '2021-04-01 13:37:00', + last_discovery_at: '2021-04-01 13:37:00', + }); + + const state = { hello: { t: 'something' } }; + + await db.transaction(tx => + db.updateEntityCache(tx, { + id, + state, + }), + ); + + const entities = await knex( + 'refresh_state', + ).select(); + expect(entities.length).toBe(1); + expect(entities[0].cache).toEqual(JSON.stringify(state)); + + await db.transaction(tx => + db.updateEntityCache(tx, { + id, + state: undefined, + }), + ); + + const entities2 = await knex( + 'refresh_state', + ).select(); + expect(entities2.length).toBe(1); + expect(entities2[0].cache).toEqual('{}'); + }, + 60_000, + ); + }); + describe('replaceUnprocessedEntities', () => { const createLocations = async (db: Knex, entityRefs: string[]) => { for (const ref of entityRefs) { diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index 263ba1bdfd..33f01b14eb 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts @@ -40,6 +40,7 @@ import { UpdateProcessedEntityOptions, ListAncestorsOptions, ListAncestorsResult, + UpdateEntityCacheOptions, } from './types'; // The number of items that are sent per batch to the database layer, when @@ -69,7 +70,6 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { id, processedEntity, resultHash, - state, errors, relations, deferredEntities, @@ -79,7 +79,6 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { .update({ processed_entity: JSON.stringify(processedEntity), result_hash: resultHash, - cache: JSON.stringify(state), errors, location_key: locationKey, }) @@ -140,6 +139,18 @@ export class DefaultProcessingDatabase implements ProcessingDatabase { .where('entity_id', id); } + async updateEntityCache( + txOpaque: Transaction, + options: UpdateEntityCacheOptions, + ): Promise { + const tx = txOpaque as Knex.Transaction; + const { id, state } = options; + + await tx('refresh_state') + .update({ cache: JSON.stringify(state ?? {}) }) + .where('entity_id', id); + } + private deduplicateRelations(rows: DbRelationsRow[]): DbRelationsRow[] { return lodash.uniqBy( rows, diff --git a/plugins/catalog-backend/src/next/database/types.ts b/plugins/catalog-backend/src/next/database/types.ts index ab449c8a7a..aabd28afc8 100644 --- a/plugins/catalog-backend/src/next/database/types.ts +++ b/plugins/catalog-backend/src/next/database/types.ts @@ -15,7 +15,7 @@ */ import { Entity, EntityRelationSpec } from '@backstage/catalog-model'; -import { JsonValue } from '@backstage/config'; +import { JsonObject } from '@backstage/config'; import { DateTime } from 'luxon'; import { Transaction } from '../../database/types'; import { DeferredEntity } from '../processing/types'; @@ -36,13 +36,17 @@ export type UpdateProcessedEntityOptions = { id: string; processedEntity: Entity; resultHash: string; - state?: JsonValue; errors?: string; relations: EntityRelationSpec[]; deferredEntities: DeferredEntity[]; locationKey?: string; }; +export type UpdateEntityCacheOptions = { + id: string; + state?: JsonObject; +}; + export type UpdateProcessedEntityErrorsOptions = { id: string; errors?: string; @@ -57,7 +61,7 @@ export type RefreshStateItem = { resultHash: string; nextUpdateAt: DateTime; lastDiscoveryAt: DateTime; // remove? - state?: JsonValue; + state?: JsonObject; errors?: string; locationKey?: string; }; @@ -118,6 +122,14 @@ export interface ProcessingDatabase { options: UpdateProcessedEntityOptions, ): Promise; + /** + * Updates the cache associated with an entity. + */ + updateEntityCache( + txOpaque: Transaction, + options: UpdateEntityCacheOptions, + ): Promise; + /** * Updates only the errors of a processed entity */ diff --git a/plugins/catalog-backend/src/next/processing/types.ts b/plugins/catalog-backend/src/next/processing/types.ts index 59a10c084f..8b533714ce 100644 --- a/plugins/catalog-backend/src/next/processing/types.ts +++ b/plugins/catalog-backend/src/next/processing/types.ts @@ -15,17 +15,17 @@ */ import { Entity, EntityRelationSpec } from '@backstage/catalog-model'; -import { JsonValue } from '@backstage/config'; +import { JsonObject } from '@backstage/config'; export type EntityProcessingRequest = { entity: Entity; - state?: JsonValue; // Versions for multiple deployments etc + state?: JsonObject; // Versions for multiple deployments etc }; export type EntityProcessingResult = | { ok: true; - state: JsonValue; + state: JsonObject; completedEntity: Entity; deferredEntities: DeferredEntity[]; relations: EntityRelationSpec[]; From b0e1aa76205fc86cb642460046921f8e74a57a7f Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Wed, 22 Sep 2021 17:35:52 +0200 Subject: [PATCH 13/15] catalog-backend: add scoped readLocation cache MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Fredrik Adelöw Co-authored-by: blam Co-authored-by: Johan Haals Signed-off-by: Patrik Oldsberg --- .../DefaultCatalogProcessingOrchestrator.ts | 2 +- .../processing/ProcessorCacheManager.test.ts | 41 +++++++++++++ .../next/processing/ProcessorCacheManager.ts | 59 +++++++++++++++++-- 3 files changed, 97 insertions(+), 5 deletions(-) diff --git a/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts b/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts index 01640f34b1..6261c0e2ff 100644 --- a/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts +++ b/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts @@ -318,7 +318,7 @@ export class DefaultCatalogProcessingOrchestrator false, context.collector.onEmit, this.options.parser, - context.cache.forProcessor(processor), + context.cache.forProcessor(processor, target), ); if (read) { didRead = true; diff --git a/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.test.ts b/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.test.ts index e7ee27b667..44c96da1ca 100644 --- a/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.test.ts +++ b/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.test.ts @@ -75,3 +75,44 @@ describe('ProcessorCacheManager', () => { ).resolves.toBe('my-new-value'); }); }); + +describe('ScopedProcessorCache', () => { + const myProcessor = new MyProcessor(); + + it('should forward existing state and collect new state', async () => { + const cache = new ProcessorCacheManager({ + 'my-processor': { 'scope-1': { 'my-key': 'my-value' } }, + }); + + const scopedCache1 = cache.forProcessor(myProcessor, 'scope-1'); + const scopedCache2 = cache.forProcessor(myProcessor, 'scope-2'); + + // Should be empty to begin with + expect(cache.collect()).toEqual({ + 'my-processor': { 'scope-1': { 'my-key': 'my-value' } }, + }); + + await scopedCache2.set('my-new-key-2', 'my-new-value-2'); + + expect(cache.collect()).toEqual({ + 'my-processor': { + 'scope-1': { 'my-key': 'my-value' }, + 'scope-2': { 'my-new-key-2': 'my-new-value-2' }, + }, + }); + + await scopedCache1.set('my-new-key', 'my-new-value'); + + await expect(scopedCache1.get('my-key')).resolves.toBe('my-value'); + await expect( + scopedCache1.get('my-new-key'), + ).resolves.toBeUndefined(); + + expect(cache.collect()).toEqual({ + 'my-processor': { + 'scope-1': { 'my-new-key': 'my-new-value' }, + 'scope-2': { 'my-new-key-2': 'my-new-value-2' }, + }, + }); + }); +}); diff --git a/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.ts b/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.ts index 5da466fe86..16945b7cc9 100644 --- a/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.ts +++ b/plugins/catalog-backend/src/next/processing/ProcessorCacheManager.ts @@ -19,7 +19,7 @@ import { CatalogProcessor } from '../../ingestion/processors'; import { CatalogProcessorCache } from '../../ingestion/processors/types'; import { isObject } from './util'; -class SingleProcessorCache implements CatalogProcessorCache { +class SingleProcessorSubCache implements CatalogProcessorCache { private newState?: JsonObject; constructor(private readonly existingState?: JsonObject) {} @@ -46,17 +46,68 @@ class SingleProcessorCache implements CatalogProcessorCache { } } +class SingleProcessorCache implements CatalogProcessorCache { + private newState?: JsonObject; + private subCaches: Map = new Map(); + + constructor(private readonly existingState?: JsonObject) {} + + async get( + key: string, + ): Promise { + return this.existingState?.[key] as ItemType | undefined; + } + + async set( + key: string, + value: ItemType, + ): Promise { + if (!this.newState) { + this.newState = {}; + } + + this.newState[key] = value; + } + + withKey(key: string) { + const existingSubCache = this.subCaches.get(key); + if (existingSubCache) { + return existingSubCache; + } + const existing = this.existingState?.[key]; + const subCache = new SingleProcessorSubCache( + isObject(existing) ? existing : undefined, + ); + this.subCaches.set(key, subCache); + return subCache; + } + + collect(): JsonObject | undefined { + let obj = this.newState ?? this.existingState; + for (const [key, subCache] of this.subCaches) { + const subCacheValue = subCache.collect(); + if (subCacheValue) { + obj = { ...obj, [key]: subCacheValue }; + } + } + return obj; + } +} + export class ProcessorCacheManager { private caches = new Map(); constructor(private readonly existingState: JsonObject) {} - forProcessor(processor: CatalogProcessor): CatalogProcessorCache { + forProcessor( + processor: CatalogProcessor, + key?: string, + ): CatalogProcessorCache { // constructor name will be deprecated in the future when we make `getProcessorName` required in the implementation const name = processor.getProcessorName?.() ?? processor.constructor.name; const cache = this.caches.get(name); if (cache) { - return cache; + return key ? cache.withKey(key) : cache; } const existing = this.existingState[name]; @@ -65,7 +116,7 @@ export class ProcessorCacheManager { isObject(existing) ? existing : undefined, ); this.caches.set(name, newCache); - return newCache; + return key ? newCache.withKey(key) : newCache; } collect(): JsonObject { From 81d3f28e71a09277cc93271b1850597bfcb3b4ac Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Wed, 22 Sep 2021 19:03:33 +0200 Subject: [PATCH 14/15] catalog-backend: update API report Signed-off-by: Patrik Oldsberg --- plugins/catalog-backend/api-report.md | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/plugins/catalog-backend/api-report.md b/plugins/catalog-backend/api-report.md index 10f149f850..133deb96fc 100644 --- a/plugins/catalog-backend/api-report.md +++ b/plugins/catalog-backend/api-report.md @@ -17,6 +17,7 @@ import { EntityPolicy } from '@backstage/catalog-model'; import { EntityRelationSpec } from '@backstage/catalog-model'; import express from 'express'; import { IndexableDocument } from '@backstage/search-common'; +import { JsonObject } from '@backstage/config'; import { JsonValue } from '@backstage/config'; import { Knex } from 'knex'; import { Location as Location_2 } from '@backstage/catalog-model'; @@ -877,7 +878,7 @@ export type EntityPagination = { // @public (undocumented) export type EntityProcessingRequest = { entity: Entity; - state?: JsonValue; + state?: JsonObject; }; // Warning: (ae-missing-release-tag) "EntityProcessingResult" is exported by the package, but it is missing a release tag (@alpha, @beta, @public, or @internal) @@ -886,7 +887,7 @@ export type EntityProcessingRequest = { export type EntityProcessingResult = | { ok: true; - state: JsonValue; + state: JsonObject; completedEntity: Entity; deferredEntities: DeferredEntity[]; relations: EntityRelationSpec[]; From 1b271e1677330d9c60762e89ff45a50557573355 Mon Sep 17 00:00:00 2001 From: Johan Haals Date: Thu, 23 Sep 2021 14:53:24 +0200 Subject: [PATCH 15/15] chore: rebase/combine tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Fredrik Adelöw Signed-off-by: Johan Haals --- ...faultCatalogProcessingOrchestrator.test.ts | 104 +++++++++--------- 1 file changed, 49 insertions(+), 55 deletions(-) diff --git a/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.test.ts b/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.test.ts index f8cf65d4e8..098f3ac14d 100644 --- a/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.test.ts +++ b/plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.test.ts @@ -18,16 +18,12 @@ import { getVoidLogger } from '@backstage/backend-common'; import { Entity, EntityPolicies, - EntityPolicy, LocationEntity, LocationSpec, LOCATION_ANNOTATION, ORIGIN_LOCATION_ANNOTATION, } from '@backstage/catalog-model'; -import { - ScmIntegrationRegistry, - ScmIntegrations, -} from '@backstage/integration'; +import { ScmIntegrations } from '@backstage/integration'; import { CatalogProcessor, CatalogProcessorCache, @@ -79,7 +75,7 @@ class FooBarProcessor implements CatalogProcessor { } describe('DefaultCatalogProcessingOrchestrator', () => { - describe('2', () => { + describe('basic processing', () => { const entity = { apiVersion: 'my-api/v1', kind: 'FooBar', @@ -92,17 +88,13 @@ describe('DefaultCatalogProcessingOrchestrator', () => { }, }; - const rulesEnforcer: CatalogRulesEnforcer = { - isAllowed: () => true, - }; - const orchestrator = new DefaultCatalogProcessingOrchestrator({ processors: [new FooBarProcessor()], integrations: ScmIntegrations.fromConfig(new ConfigReader({})), logger: getVoidLogger(), parser: defaultEntityDataParser, policy: EntityPolicies.allOf([]), - rulesEnforcer, + rulesEnforcer: { isAllowed: () => true }, }); it('runs a minimal processing', async () => { @@ -196,53 +188,55 @@ describe('DefaultCatalogProcessingOrchestrator', () => { }); }); - it('enforces catalog rules', async () => { - const entity: LocationEntity = { - apiVersion: 'backstage.io/v1beta1', - kind: 'Location', - metadata: { - name: 'l', - annotations: { - [ORIGIN_LOCATION_ANNOTATION]: 'url:https://example.com/origin.yaml', - [LOCATION_ANNOTATION]: 'url:https://example.com/origin.yaml', + describe('rules', () => { + it('enforces catalog rules', async () => { + const entity: LocationEntity = { + apiVersion: 'backstage.io/v1beta1', + kind: 'Location', + metadata: { + name: 'l', + annotations: { + [ORIGIN_LOCATION_ANNOTATION]: 'url:https://example.com/origin.yaml', + [LOCATION_ANNOTATION]: 'url:https://example.com/origin.yaml', + }, }, - }, - spec: { - type: 'url', - target: 'http://example.com/entity.yaml', - }, - }; + spec: { + type: 'url', + target: 'http://example.com/entity.yaml', + }, + }; - const integrations = ScmIntegrations.fromConfig(new ConfigReader({})); - const processor: jest.Mocked = { - validateEntityKind: jest.fn(async () => true), - readLocation: jest.fn(async (_l, _o, emit) => { - emit(results.entity({ type: 't', target: 't' }, entity)); - return true; - }), - }; - const parser: CatalogProcessorParser = jest.fn(); - const rulesEnforcer: jest.Mocked = { - isAllowed: jest.fn(), - }; + const integrations = ScmIntegrations.fromConfig(new ConfigReader({})); + const processor: jest.Mocked = { + validateEntityKind: jest.fn(async () => true), + readLocation: jest.fn(async (_l, _o, emit) => { + emit(results.entity({ type: 't', target: 't' }, entity)); + return true; + }), + }; + const parser: CatalogProcessorParser = jest.fn(); + const rulesEnforcer: jest.Mocked = { + isAllowed: jest.fn(), + }; - const orchestrator = new DefaultCatalogProcessingOrchestrator({ - processors: [processor], - integrations, - logger: getVoidLogger(), - parser, - policy: EntityPolicies.allOf([]), - rulesEnforcer, + const orchestrator = new DefaultCatalogProcessingOrchestrator({ + processors: [processor], + integrations, + logger: getVoidLogger(), + parser, + policy: EntityPolicies.allOf([]), + rulesEnforcer, + }); + + rulesEnforcer.isAllowed.mockReturnValueOnce(true); + await expect( + orchestrator.process({ entity, state: {} }), + ).resolves.toEqual(expect.objectContaining({ ok: true })); + + rulesEnforcer.isAllowed.mockReturnValueOnce(false); + await expect( + orchestrator.process({ entity, state: {} }), + ).resolves.toEqual(expect.objectContaining({ ok: false })); }); - - rulesEnforcer.isAllowed.mockReturnValueOnce(true); - await expect(orchestrator.process({ entity, state: {} })).resolves.toEqual( - expect.objectContaining({ ok: true }), - ); - - rulesEnforcer.isAllowed.mockReturnValueOnce(false); - await expect(orchestrator.process({ entity, state: {} })).resolves.toEqual( - expect.objectContaining({ ok: false }), - ); }); });