From 33333c94169e1d349f0568516c74143e7d9ddd9a Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Mon, 20 Sep 2021 16:14:45 +0200 Subject: [PATCH] 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[];