From d94b7c3dce9c0f243bca596265ce818b58781a10 Mon Sep 17 00:00:00 2001 From: Patrik Oldsberg Date: Mon, 20 Sep 2021 13:49:53 +0200 Subject: [PATCH] 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 = {