From e19a93ee5594ba9402f58fd38efc3193a6926e4b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Sat, 6 Jun 2020 23:36:30 +0200 Subject: [PATCH] Get rid of the generators --- .../src/database/CommonDatabase.ts | 7 +- .../src/ingestion/HigherOrderOperations.ts | 9 +- .../src/ingestion/LocationReaders.ts | 255 +++++++++--------- .../AnnotateLocationEntityProcessor.ts | 11 +- .../processors/EntityPolicyProcessor.ts | 17 +- .../processors/FileReaderProcessor.ts | 28 +- .../processors/GithubReaderProcessor.ts | 29 +- .../src/ingestion/processors/YamlProcessor.ts | 21 +- .../src/ingestion/processors/types.ts | 85 +++++- 9 files changed, 253 insertions(+), 209 deletions(-) diff --git a/plugins/catalog-backend/src/database/CommonDatabase.ts b/plugins/catalog-backend/src/database/CommonDatabase.ts index aeea89aef3..882995c142 100644 --- a/plugins/catalog-backend/src/database/CommonDatabase.ts +++ b/plugins/catalog-backend/src/database/CommonDatabase.ts @@ -19,12 +19,7 @@ import { InputError, NotFoundError, } from '@backstage/backend-common'; -import { - Entity, - EntityMeta, - Location, - LOCATION_ANNOTATION, -} from '@backstage/catalog-model'; +import { Entity, EntityMeta, Location } from '@backstage/catalog-model'; import Knex from 'knex'; import lodash from 'lodash'; import { v4 as uuidv4 } from 'uuid'; diff --git a/plugins/catalog-backend/src/ingestion/HigherOrderOperations.ts b/plugins/catalog-backend/src/ingestion/HigherOrderOperations.ts index bc346f8a4d..2080b1c878 100644 --- a/plugins/catalog-backend/src/ingestion/HigherOrderOperations.ts +++ b/plugins/catalog-backend/src/ingestion/HigherOrderOperations.ts @@ -15,21 +15,16 @@ */ import { InputError } from '@backstage/backend-common'; -import { - Entity, - Location, - LocationSpec, - LOCATION_ANNOTATION, -} from '@backstage/catalog-model'; +import { Entity, Location, LocationSpec } from '@backstage/catalog-model'; import lodash from 'lodash'; import { v4 as uuidv4 } from 'uuid'; +import { Logger } from 'winston'; import { EntitiesCatalog, LocationsCatalog } from '../catalog'; import { AddLocationResult, HigherOrderOperation, LocationReader, } from './types'; -import { Logger } from 'winston'; /** * Placeholder for operations that span several catalogs and/or stretches out diff --git a/plugins/catalog-backend/src/ingestion/LocationReaders.ts b/plugins/catalog-backend/src/ingestion/LocationReaders.ts index a40e86d177..708c3e11aa 100644 --- a/plugins/catalog-backend/src/ingestion/LocationReaders.ts +++ b/plugins/catalog-backend/src/ingestion/LocationReaders.ts @@ -25,10 +25,15 @@ import { AnnotateLocationEntityProcessor } from './processors/AnnotateLocationEn import { EntityPolicyProcessor } from './processors/EntityPolicyProcessor'; import { FileReaderProcessor } from './processors/FileReaderProcessor'; import { GithubReaderProcessor } from './processors/GithubReaderProcessor'; +import * as result from './processors/results'; import { LocationProcessor, + LocationProcessorDataResult, + LocationProcessorEntityResult, + LocationProcessorErrorResult, + LocationProcessorLocationResult, LocationProcessorResult, - LocationProcessorResults, + LocationProcessorSink, } from './processors/types'; import { YamlProcessor } from './processors/YamlProcessor'; import { LocationReader, ReadLocationResult } from './types'; @@ -65,143 +70,137 @@ export class LocationReaders implements LocationReader { async read(location: LocationSpec): Promise { const output: ReadLocationResult = { entities: [], errors: [] }; - const initialItem: LocationProcessorResult = { - type: 'location', - location, - optional: false, - }; - await this.handleResultItem(initialItem, 0, output); + let items: LocationProcessorResult[] = [result.location(location, false)]; + + for (let depth = 0; depth < MAX_DEPTH; ++depth) { + const newItems: LocationProcessorResult[] = []; + const sink: LocationProcessorSink = i => newItems.push(i); + + for (const item of items) { + if (item.type === 'location') { + await this.handleLocation(item, sink); + } else if (item.type === 'data') { + await this.handleData(item, sink); + } else if (item.type === 'entity') { + await this.handleEntity(item, sink, output); + } else if (item.type === 'error') { + await this.handleError(item, sink, output); + } + } + + if (newItems.length === 0) { + return output; + } + + items = newItems; + } + + const message = `Max recursion depth ${MAX_DEPTH} reached for ${location.type} ${location.target}`; + this.logger.warn(message); + output.errors.push({ location, error: new Error(message) }); return output; } - async handleResultItem( - item: LocationProcessorResult, - depth: number, - output: ReadLocationResult, - ): Promise { - // Sanity check to break silly expansions / loops - if (depth > MAX_DEPTH) { - const message = `Max recursion depth ${MAX_DEPTH} reached at ${item.location.type} ${item.location.target}`; - this.logger.warn(message); - output.errors.push({ - location: item.location, - error: new Error(message), - }); - return; - } + private async handleLocation( + item: LocationProcessorLocationResult, + emit: LocationProcessorSink, + ) { + this.logger.debug( + `Reading location ${item.location.type} ${item.location.target} optional=${item.optional}`, + ); - if (item.type === 'location') { - this.logger.debug( - `Reading location ${item.location.type} ${item.location.target} optional=${item.optional}`, - ); - await this.runAll( - 'fetch', - processor => processor.readLocation?.(item.location, item.optional), - emitted => this.handleResultItem(emitted, depth + 1, output), - item.location, - true, - true, - ); - } else if (item.type === 'data') { - this.logger.debug( - `Parsing data from location ${item.location.type} ${item.location.target} (${item.data.byteLength} bytes)`, - ); - await this.runAll( - 'parse', - processor => processor.parseData?.(item.data, item.location), - emitted => this.handleResultItem(emitted, depth + 1, output), - item.location, - true, - true, - ); - } else if (item.type === 'error') { - this.logger.debug( - `Encountered error at location ${item.location.type} ${item.location.target}, ${item.error}`, - ); - await this.runAll( - 'process error', - processor => processor.handleError?.(item.error, item.location), - emitted => this.handleResultItem(emitted, depth + 1, output), - item.location, - false, - false, - ); - output.errors.push({ - location: item.location, - error: item.error, - }); - } else if (item.type === 'entity') { - this.logger.debug( - `Got entity at location ${item.location.type} ${item.location.target}, ${item.entity.apiVersion} ${item.entity.kind}`, - ); - const current = { entity: item.entity, location: item.location }; - await this.runAll( - 'process entity', - processor => - processor.processEntity?.(current.entity, current.location), - async emitted => { - if (emitted.type === 'entity') { - current.entity = emitted.entity; - current.location = emitted.location; - } else { - await this.handleResultItem(emitted, depth + 1, output); - } - }, - item.location, - false, - false, - ); - output.entities.push({ - entity: current.entity, - location: current.location, - }); - } - } - - async runAll( - what: string, - start: ( - processor: LocationProcessor, - ) => LocationProcessorResults | undefined, - emit: (item: LocationProcessorResult) => Promise, - location: LocationSpec, - stopAfterFirstHandled: boolean, - failIfNotHandled: boolean, - ): Promise { - let wasHandled = false; for (const processor of this.processors) { - try { - const iterator = start(processor); - if (!iterator) { - continue; - } - - for (;;) { - const item = await iterator.next(); - if (item.done) { - break; + if (processor.readLocation) { + try { + if ( + await processor.readLocation(item.location, item.optional, emit) + ) { + return; } - if (!item.value) { - continue; - } - - wasHandled = true; - await emit(item.value); + } catch (e) { + const message = `Processor ${processor.constructor.name} threw an error while reading location ${item.location.type} ${item.location.target}, ${e}`; + emit(result.generalError(item.location, message)); } - - if (wasHandled && stopAfterFirstHandled) { - return; - } - } catch (e) { - const message = `Processor ${processor.constructor.name} threw an error during ${what}, ${e}`; - await emit({ type: 'error', location, error: new Error(message) }); - return; } } - if (!wasHandled && failIfNotHandled) { - const message = `No processor was able to handle ${location.type} ${location.target} during ${what}`; - await emit({ type: 'error', location, error: new Error(message) }); + const message = `No processor was able to read location ${item.location.type} ${item.location.target}`; + emit(result.inputError(item.location, message)); + } + + private async handleData( + item: LocationProcessorDataResult, + emit: LocationProcessorSink, + ) { + this.logger.debug( + `Parsing data from location ${item.location.type} ${item.location.target} (${item.data.byteLength} bytes)`, + ); + + for (const processor of this.processors) { + if (processor.parseData) { + try { + if (await processor.parseData(item.data, item.location, emit)) { + return; + } + } catch (e) { + const message = `Processor ${processor.constructor.name} threw an error while parsing ${item.location.type} ${item.location.target}, ${e}`; + emit(result.generalError(item.location, message)); + } + } } + + const message = `No processor was able to parse location ${item.location.type} ${item.location.target}`; + emit(result.inputError(item.location, message)); + } + + private async handleEntity( + item: LocationProcessorEntityResult, + emit: LocationProcessorSink, + output: ReadLocationResult, + ) { + this.logger.debug( + `Got entity at location ${item.location.type} ${item.location.target}, ${item.entity.apiVersion} ${item.entity.kind}`, + ); + + let current = item.entity; + + for (const processor of this.processors) { + if (processor.processEntity) { + try { + current = await processor.processEntity(current, item.location, emit); + } catch (e) { + const message = `Processor ${processor.constructor.name} threw an error while processing entity at ${item.location.type} ${item.location.target}, ${e}`; + emit(result.generalError(item.location, message)); + } + } + } + + output.entities.push({ entity: current, location: item.location }); + } + + private async handleError( + item: LocationProcessorErrorResult, + emit: LocationProcessorSink, + output: ReadLocationResult, + ) { + this.logger.debug( + `Encountered error at location ${item.location.type} ${item.location.target}, ${item.error}`, + ); + + for (const processor of this.processors) { + if (processor.handleError) { + try { + await processor.handleError(item.error, item.location, emit); + } catch (e) { + const message = `Processor ${processor.constructor.name} threw an error while handling another error at ${item.location.type} ${item.location.target}, ${e}`; + emit(result.generalError(item.location, message)); + } + } + } + + output.errors.push({ + location: item.location, + error: item.error, + }); } } diff --git a/plugins/catalog-backend/src/ingestion/processors/AnnotateLocationEntityProcessor.ts b/plugins/catalog-backend/src/ingestion/processors/AnnotateLocationEntityProcessor.ts index e0cf90c444..ec7e59281e 100644 --- a/plugins/catalog-backend/src/ingestion/processors/AnnotateLocationEntityProcessor.ts +++ b/plugins/catalog-backend/src/ingestion/processors/AnnotateLocationEntityProcessor.ts @@ -16,15 +16,11 @@ import { Entity, LocationSpec } from '@backstage/catalog-model'; import lodash from 'lodash'; -import { LocationProcessor, LocationProcessorResults } from './types'; -import * as result from './results'; +import { LocationProcessor } from './types'; export class AnnotateLocationEntityProcessor implements LocationProcessor { - async *processEntity( - entity: Entity, - location: LocationSpec, - ): LocationProcessorResults { - const merged = lodash.merge( + async processEntity(entity: Entity, location: LocationSpec): Promise { + return lodash.merge( { metadata: { annotations: { @@ -34,6 +30,5 @@ export class AnnotateLocationEntityProcessor implements LocationProcessor { }, entity, ); - yield result.entity(location, merged); } } diff --git a/plugins/catalog-backend/src/ingestion/processors/EntityPolicyProcessor.ts b/plugins/catalog-backend/src/ingestion/processors/EntityPolicyProcessor.ts index 5343f133ae..7360a01a84 100644 --- a/plugins/catalog-backend/src/ingestion/processors/EntityPolicyProcessor.ts +++ b/plugins/catalog-backend/src/ingestion/processors/EntityPolicyProcessor.ts @@ -14,9 +14,8 @@ * limitations under the License. */ -import { Entity, EntityPolicy, LocationSpec } from '@backstage/catalog-model'; -import * as result from './results'; -import { LocationProcessor, LocationProcessorResults } from './types'; +import { Entity, EntityPolicy } from '@backstage/catalog-model'; +import { LocationProcessor } from './types'; export class EntityPolicyProcessor implements LocationProcessor { private readonly policy: EntityPolicy; @@ -25,15 +24,7 @@ export class EntityPolicyProcessor implements LocationProcessor { this.policy = policy; } - async *processEntity( - entity: Entity, - location: LocationSpec, - ): LocationProcessorResults { - try { - const updatedEntity = await this.policy.enforce(entity); - yield result.entity(location, updatedEntity); - } catch (e) { - yield result.generalError(location, e.toString()); - } + async processEntity(entity: Entity): Promise { + return await this.policy.enforce(entity); } } diff --git a/plugins/catalog-backend/src/ingestion/processors/FileReaderProcessor.ts b/plugins/catalog-backend/src/ingestion/processors/FileReaderProcessor.ts index 9263b0893c..a3d24575bc 100644 --- a/plugins/catalog-backend/src/ingestion/processors/FileReaderProcessor.ts +++ b/plugins/catalog-backend/src/ingestion/processors/FileReaderProcessor.ts @@ -17,32 +17,32 @@ import { LocationSpec } from '@backstage/catalog-model'; import fs from 'fs-extra'; import * as result from './results'; -import { LocationProcessor, LocationProcessorResults } from './types'; +import { LocationProcessor, LocationProcessorSink } from './types'; export class FileReaderProcessor implements LocationProcessor { - async *readLocation( + async readLocation( location: LocationSpec, optional: boolean, - ): LocationProcessorResults { + emit: LocationProcessorSink, + ): Promise { if (location.type !== 'file') { - return; + return false; } try { const exists = await fs.pathExists(location.target); - if (!exists) { - if (!optional) { - const message = `${location.type} ${location.target} does not exist`; - yield result.notFoundError(location, message); - } - return; + if (exists) { + const data = await fs.readFile(location.target); + emit(result.data(location, data)); + } else if (!optional) { + const message = `${location.type} ${location.target} does not exist`; + emit(result.notFoundError(location, message)); } - - const data = await fs.readFile(location.target); - yield result.data(location, data); } catch (e) { const message = `${location.type} ${location.target} could not be read, ${e}`; - yield result.generalError(location, message); + emit(result.generalError(location, message)); } + + return true; } } diff --git a/plugins/catalog-backend/src/ingestion/processors/GithubReaderProcessor.ts b/plugins/catalog-backend/src/ingestion/processors/GithubReaderProcessor.ts index 00e3277037..a39380027b 100644 --- a/plugins/catalog-backend/src/ingestion/processors/GithubReaderProcessor.ts +++ b/plugins/catalog-backend/src/ingestion/processors/GithubReaderProcessor.ts @@ -17,12 +17,16 @@ import { LocationSpec } from '@backstage/catalog-model'; import fetch from 'node-fetch'; import * as result from './results'; -import { LocationProcessor, LocationProcessorResults } from './types'; +import { LocationProcessor, LocationProcessorSink } from './types'; export class GithubReaderProcessor implements LocationProcessor { - async *readLocation(location: LocationSpec): LocationProcessorResults { + async readLocation( + location: LocationSpec, + optional: boolean, + emit: LocationProcessorSink, + ): Promise { if (location.type !== 'github') { - return; + return false; } try { @@ -32,22 +36,25 @@ export class GithubReaderProcessor implements LocationProcessor { // notFound instead of fatal? const response = await fetch(url.toString()); - if (!response.ok) { + if (response.ok) { + const data = await response.buffer(); + emit(result.data(location, data)); + } else { const message = `${location.target} could not be read as ${url}, ${response.status} ${response.statusText}`; if (response.status === 404) { - yield result.notFoundError(location, message); + if (!optional) { + emit(result.notFoundError(location, message)); + } } else { - yield result.generalError(location, message); + emit(result.generalError(location, message)); } - return; } - - const data = await response.buffer(); - yield result.data(location, data); } catch (e) { const message = `Unable to read ${location.type} ${location.target}, ${e}`; - yield result.generalError(location, message); + emit(result.generalError(location, message)); } + + return true; } // Converts diff --git a/plugins/catalog-backend/src/ingestion/processors/YamlProcessor.ts b/plugins/catalog-backend/src/ingestion/processors/YamlProcessor.ts index 291141c546..1faf77646a 100644 --- a/plugins/catalog-backend/src/ingestion/processors/YamlProcessor.ts +++ b/plugins/catalog-backend/src/ingestion/processors/YamlProcessor.ts @@ -17,39 +17,42 @@ import { Entity, LocationSpec } from '@backstage/catalog-model'; import lodash from 'lodash'; import yaml from 'yaml'; -import { LocationProcessor, LocationProcessorResults } from './types'; import * as result from './results'; +import { LocationProcessor, LocationProcessorSink } from './types'; export class YamlProcessor implements LocationProcessor { - async *parseData( + async parseData( data: Buffer, location: LocationSpec, - ): LocationProcessorResults { + emit: LocationProcessorSink, + ): Promise { if (!location.target.match(/\.ya?ml$/)) { - return; + return false; } let documents: yaml.Document.Parsed[]; try { documents = yaml.parseAllDocuments(data.toString('utf8')).filter(d => d); } catch (e) { - yield result.generalError(location, `Failed to parse YAML, ${e}`); - return; + emit(result.generalError(location, `Failed to parse YAML, ${e}`)); + return true; } for (const document of documents) { if (document.errors?.length) { const message = `YAML error, ${document.errors[0]}`; - yield result.generalError(location, message); + emit(result.generalError(location, message)); } else { const json = document.toJSON(); if (lodash.isPlainObject(json)) { - yield result.entity(location, json as Entity); + emit(result.entity(location, json as Entity)); } else { const message = `Expected object at root, got ${typeof json}`; - yield result.generalError(location, message); + emit(result.generalError(location, message)); } } } + + return true; } } diff --git a/plugins/catalog-backend/src/ingestion/processors/types.ts b/plugins/catalog-backend/src/ingestion/processors/types.ts index 0be7382c2a..43e5480029 100644 --- a/plugins/catalog-backend/src/ingestion/processors/types.ts +++ b/plugins/catalog-backend/src/ingestion/processors/types.ts @@ -21,30 +21,89 @@ export type LocationProcessor = { * 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 + * @returns True if handled by this processor, false otherwise */ readLocation?( location: LocationSpec, optional: boolean, - ): LocationProcessorResults; + emit: LocationProcessorSink, + ): Promise; - parseData?(data: Buffer, location: LocationSpec): LocationProcessorResults; + /** + * Parses a raw data buffer that was read from a location. + * + * @param data The data to parse + * @param location The location that the data came from + * @param emit A sink for items resulting from the parsing + * @returns True if handled by this processor, false otherwise + */ + parseData?( + data: Buffer, + location: LocationSpec, + emit: LocationProcessorSink, + ): Promise; + /** + * Processes an emitted entity, e.g. by validating or modifying it. + * + * @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 + * @returns The same entity or a modifid version of it + */ processEntity?( entity: Entity, location: LocationSpec, - ): LocationProcessorResults; + emit: LocationProcessorSink, + ): Promise; - handleError?(error: Error, location: LocationSpec): LocationProcessorResults; + /** + * 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 handilng + * @returns Nothing + */ + handleError?( + error: Error, + location: LocationSpec, + emit: LocationProcessorSink, + ): Promise; }; -export type LocationProcessorResults = AsyncGenerator< - LocationProcessorResult, - void, - unknown ->; +export type LocationProcessorSink = ( + generated: LocationProcessorResult, +) => void; + +export type LocationProcessorLocationResult = { + type: 'location'; + location: LocationSpec; + optional: boolean; +}; + +export type LocationProcessorDataResult = { + type: 'data'; + data: Buffer; + location: LocationSpec; +}; + +export type LocationProcessorEntityResult = { + type: 'entity'; + entity: Entity; + location: LocationSpec; +}; + +export type LocationProcessorErrorResult = { + type: 'error'; + error: Error; + location: LocationSpec; +}; export type LocationProcessorResult = - | { type: 'error'; error: Error; location: LocationSpec } // An error occurred - | { type: 'location'; location: LocationSpec; optional: boolean } // A location to read - | { type: 'data'; data: Buffer; location: LocationSpec } // Some raw data was read - | { type: 'entity'; entity: Entity; location: LocationSpec }; // An entity was produced + | LocationProcessorLocationResult + | LocationProcessorDataResult + | LocationProcessorEntityResult + | LocationProcessorErrorResult;