From d68e4f2bd001ce53dee344300b2ee36bbe75d6b3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Fri, 5 Jun 2020 09:16:55 +0200 Subject: [PATCH] Switch to using generators --- packages/backend/src/plugins/catalog.ts | 4 +- .../src/database/CommonDatabase.ts | 6 - .../src/ingestion/LocationReaders.ts | 263 ++++++++---------- .../AnnotateLocationEntityProcessor.ts | 23 +- .../processors/EntityPolicyProcessor.ts | 17 +- .../processors/FileReaderProcessor.ts | 29 +- .../processors/GithubReaderProcessor.ts | 44 +-- .../src/ingestion/processors/YamlProcessor.ts | 37 +-- .../src/ingestion/processors/results.ts | 69 +++++ .../src/ingestion/processors/types.ts | 27 +- .../src/service/standaloneServer.ts | 16 +- 11 files changed, 295 insertions(+), 240 deletions(-) create mode 100644 plugins/catalog-backend/src/ingestion/processors/results.ts diff --git a/packages/backend/src/plugins/catalog.ts b/packages/backend/src/plugins/catalog.ts index 26f18d6b95..ccd5d06908 100644 --- a/packages/backend/src/plugins/catalog.ts +++ b/packages/backend/src/plugins/catalog.ts @@ -29,7 +29,7 @@ export default async function createPlugin({ logger, database, }: PluginEnvironment) { - const ingestionModel = new LocationReaders(); + const locationReader = new LocationReaders(); const db = await DatabaseManager.createDatabase(database, logger); const entitiesCatalog = new DatabaseEntitiesCatalog(db); @@ -37,7 +37,7 @@ export default async function createPlugin({ const higherOrderOperation = new HigherOrderOperations( entitiesCatalog, locationsCatalog, - ingestionModel, + locationReader, logger, ); diff --git a/plugins/catalog-backend/src/database/CommonDatabase.ts b/plugins/catalog-backend/src/database/CommonDatabase.ts index bb5b95d9ba..aeea89aef3 100644 --- a/plugins/catalog-backend/src/database/CommonDatabase.ts +++ b/plugins/catalog-backend/src/database/CommonDatabase.ts @@ -169,12 +169,6 @@ export class CommonDatabase implements Database { uid: generateUid(), etag: generateEtag(), generation: 1, - annotations: { - ...(newEntity.metadata?.annotations ?? {}), - ...(request.locationId - ? { [LOCATION_ANNOTATION]: request.locationId } - : {}), - }, }; const newRow = toEntityRow(request.locationId, newEntity); diff --git a/plugins/catalog-backend/src/ingestion/LocationReaders.ts b/plugins/catalog-backend/src/ingestion/LocationReaders.ts index 92c11d1138..92f1a235ee 100644 --- a/plugins/catalog-backend/src/ingestion/LocationReaders.ts +++ b/plugins/catalog-backend/src/ingestion/LocationReaders.ts @@ -14,9 +14,7 @@ * limitations under the License. */ -import { NotFoundError } from '@backstage/backend-common'; import { - Entity, EntityPolicies, EntityPolicy, LocationSpec, @@ -25,14 +23,16 @@ import { AnnotateLocationEntityProcessor } from './processors/AnnotateLocationEn import { EntityPolicyProcessor } from './processors/EntityPolicyProcessor'; import { FileReaderProcessor } from './processors/FileReaderProcessor'; import { GithubReaderProcessor } from './processors/GithubReaderProcessor'; -import { LocationProcessor, LocationProcessorResult } from './processors/types'; +import { + LocationProcessor, + LocationProcessorResult, + LocationProcessorResults, +} from './processors/types'; import { YamlProcessor } from './processors/YamlProcessor'; import { LocationReader, ReadLocationResult } from './types'; // The max amount of nesting depth of generated work items -const MAX_DEPTH = 5; - -type QueueItem = LocationProcessorResult & { depth: number }; +const MAX_DEPTH = 10; /** * Implements the reading of a location through a series of processor tasks. @@ -59,157 +59,122 @@ export class LocationReaders implements LocationReader { } async read(location: LocationSpec): Promise { - const result: ReadLocationResult = { entities: [], errors: [] }; - - const queue: QueueItem[] = []; - queue.push({ type: 'location', location, optional: false, depth: 0 }); - - while (queue.length) { - const entry = queue.shift()!; - const depth = entry.depth + 1; - - if (depth > MAX_DEPTH) { - throw new Error( - `Failed to read ${location.type} ${location.target}, max depth exceeded`, - ); - } - - if (entry.type === 'location') { - await this.handleLocation(entry.location, entry.optional, depth, queue); - } else if (entry.type === 'data') { - await this.handleData(entry.data, entry.location, depth, queue); - } else if (entry.type === 'error') { - await this.handleError(entry.error, entry.location, depth, result); - } else if (entry.type === 'entity') { - await this.handleEntity( - entry.entity, - entry.location, - depth, - queue, - result, - ); - } - } - - return result; - } - - async handleLocation( - location: LocationSpec, - optional: boolean, - depth: number, - queue: QueueItem[], - ): Promise { - for (const processor of this.processors) { - try { - const processorOutput = await processor.readLocation?.(location); - if (processorOutput) { - processorOutput.forEach(r => queue.push({ ...r, depth })); - return; - } - } catch (e) { - if (!(e instanceof NotFoundError && optional)) { - queue.push({ - type: 'error', - error: e, - location, - depth, - }); - } - } - } - - queue.push({ - type: 'error', + const output: ReadLocationResult = { entities: [], errors: [] }; + const initialItem: LocationProcessorResult = { + type: 'location', location, - depth, - error: new Error( - `No processor could read location ${location.type} ${location.target}`, - ), - }); + optional: false, + }; + await this.handleResultItem(initialItem, 0, output); + return output; } - async handleData( - data: Buffer, - location: LocationSpec, + async handleResultItem( + item: LocationProcessorResult, depth: number, - queue: QueueItem[], + output: ReadLocationResult, ): Promise { - for (const processor of this.processors) { - try { - const processorOutput = await processor.parseData?.(data, location); - if (processorOutput) { - processorOutput.forEach(r => queue.push({ ...r, depth })); - return; - } - } catch (e) { - queue.push({ type: 'error', location, error: e, depth }); - return; - } + // Sanity check to break silly expansions / loops + if (depth > MAX_DEPTH) { + output.errors.push({ + location: item.location, + error: new Error(`Max recursion depth ${MAX_DEPTH} reached`), + }); + return; } - queue.push({ - type: 'error', - location, - depth, - error: new Error( - `No processor could parse location ${location.type} ${location.target}`, - ), - }); - } - - async handleError( - error: Error, - location: LocationSpec, - _depth: number, - result: ReadLocationResult, - ): Promise { - for (const processor of this.processors) { - try { - await processor.handleError?.(error, location); - } catch { - // ignore - } - } - - result.errors.push({ location, error }); - } - - async handleEntity( - entity: Entity, - location: LocationSpec, - depth: number, - queue: QueueItem[], - result: ReadLocationResult, - ): Promise { - let resultingEntity = entity; - let foundErrors = false; - - for (const processor of this.processors) { - try { - const processorOutput = await processor.processEntity?.( - entity, - location, - ); - if (processorOutput) { - resultingEntity = processorOutput; - } - } catch (e) { - foundErrors = true; - queue.push({ - type: 'error', - location, - error: e, - depth, - }); - } - } - - if (!foundErrors) { - result.entities.push({ - location, - entity: resultingEntity, + if (item.type === 'location') { + await this.runAll( + processor => processor.readLocation?.(item.location, item.optional), + emitted => this.handleResultItem(emitted, depth + 1, output), + item.location, + true, + true, + ); + } else if (item.type === 'data') { + await this.runAll( + processor => processor.parseData?.(item.data, item.location), + emitted => this.handleResultItem(emitted, depth + 1, output), + item.location, + true, + true, + ); + } else if (item.type === 'error') { + await this.runAll( + 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') { + const current = { entity: item.entity, location: item.location }; + await this.runAll( + 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( + 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; + } + + wasHandled = true; + await emit(item.value); + } + + if (wasHandled && stopAfterFirstHandled) { + return; + } + } catch (e) { + const message = `Processor ${processor.constructor.name} threw an error, ${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}`; + await emit({ type: 'error', location, error: new Error(message) }); + } + } + } } diff --git a/plugins/catalog-backend/src/ingestion/processors/AnnotateLocationEntityProcessor.ts b/plugins/catalog-backend/src/ingestion/processors/AnnotateLocationEntityProcessor.ts index fac6298660..e0cf90c444 100644 --- a/plugins/catalog-backend/src/ingestion/processors/AnnotateLocationEntityProcessor.ts +++ b/plugins/catalog-backend/src/ingestion/processors/AnnotateLocationEntityProcessor.ts @@ -16,13 +16,24 @@ import { Entity, LocationSpec } from '@backstage/catalog-model'; import lodash from 'lodash'; -import { LocationProcessor } from './types'; +import { LocationProcessor, LocationProcessorResults } from './types'; +import * as result from './results'; export class AnnotateLocationEntityProcessor implements LocationProcessor { - async processEntity(entity: Entity, location: LocationSpec): Promise { - const annotations = { - 'backstage.io/managed-by-location': `${location.type}:${location.target}`, - }; - return lodash.merge({ metadata: { annotations } }, entity); + async *processEntity( + entity: Entity, + location: LocationSpec, + ): LocationProcessorResults { + const merged = lodash.merge( + { + metadata: { + annotations: { + 'backstage.io/managed-by-location': `${location.type}:${location.target}`, + }, + }, + }, + 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 fcc7c71578..5343f133ae 100644 --- a/plugins/catalog-backend/src/ingestion/processors/EntityPolicyProcessor.ts +++ b/plugins/catalog-backend/src/ingestion/processors/EntityPolicyProcessor.ts @@ -14,8 +14,9 @@ * limitations under the License. */ -import { Entity, EntityPolicy } from '@backstage/catalog-model'; -import { LocationProcessor } from './types'; +import { Entity, EntityPolicy, LocationSpec } from '@backstage/catalog-model'; +import * as result from './results'; +import { LocationProcessor, LocationProcessorResults } from './types'; export class EntityPolicyProcessor implements LocationProcessor { private readonly policy: EntityPolicy; @@ -24,7 +25,15 @@ export class EntityPolicyProcessor implements LocationProcessor { this.policy = policy; } - async processEntity(entity: Entity): Promise { - return this.policy.enforce(entity); + 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()); + } } } diff --git a/plugins/catalog-backend/src/ingestion/processors/FileReaderProcessor.ts b/plugins/catalog-backend/src/ingestion/processors/FileReaderProcessor.ts index 6aa9100506..9263b0893c 100644 --- a/plugins/catalog-backend/src/ingestion/processors/FileReaderProcessor.ts +++ b/plugins/catalog-backend/src/ingestion/processors/FileReaderProcessor.ts @@ -14,28 +14,35 @@ * limitations under the License. */ -import { NotFoundError } from '@backstage/backend-common'; import { LocationSpec } from '@backstage/catalog-model'; import fs from 'fs-extra'; -import { LocationProcessor, LocationProcessorResult } from './types'; +import * as result from './results'; +import { LocationProcessor, LocationProcessorResults } from './types'; export class FileReaderProcessor implements LocationProcessor { - async readLocation( + async *readLocation( location: LocationSpec, - ): Promise { + optional: boolean, + ): LocationProcessorResults { if (location.type !== 'file') { - return undefined; - } - - if (!(await fs.pathExists(location.target))) { - throw new NotFoundError(`${location.target} does not exist`); + return; } 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; + } + const data = await fs.readFile(location.target); - return [{ type: 'data', location, data }]; + yield result.data(location, data); } catch (e) { - throw new Error(`Unable to read ${location.target}, ${e}`); + const message = `${location.type} ${location.target} could not be read, ${e}`; + yield result.generalError(location, message); } } } diff --git a/plugins/catalog-backend/src/ingestion/processors/GithubReaderProcessor.ts b/plugins/catalog-backend/src/ingestion/processors/GithubReaderProcessor.ts index 1533f0a633..00e3277037 100644 --- a/plugins/catalog-backend/src/ingestion/processors/GithubReaderProcessor.ts +++ b/plugins/catalog-backend/src/ingestion/processors/GithubReaderProcessor.ts @@ -14,35 +14,39 @@ * limitations under the License. */ -import { NotFoundError } from '@backstage/backend-common'; import { LocationSpec } from '@backstage/catalog-model'; import fetch from 'node-fetch'; -import { LocationProcessor, LocationProcessorResult } from './types'; +import * as result from './results'; +import { LocationProcessor, LocationProcessorResults } from './types'; export class GithubReaderProcessor implements LocationProcessor { - async readLocation( - location: LocationSpec, - ): Promise { + async *readLocation(location: LocationSpec): LocationProcessorResults { if (location.type !== 'github') { - return undefined; - } - - const url = this.buildRawUrl(location.target); - const response = await fetch(url.toString()); // May also throw - - if (!response.ok) { - const message = `${location.target} could not be read as ${url}, ${response.status} ${response.statusText}`; - if (response.status === 404) { - throw new NotFoundError(message); - } else { - throw new Error(message); - } + return; } try { - return [{ type: 'data', location, data: await response.buffer() }]; + const url = this.buildRawUrl(location.target); + + // TODO(freben): Should "hard" errors thrown by this line be treated as + // notFound instead of fatal? + const response = await fetch(url.toString()); + + if (!response.ok) { + const message = `${location.target} could not be read as ${url}, ${response.status} ${response.statusText}`; + if (response.status === 404) { + yield result.notFoundError(location, message); + } else { + yield result.generalError(location, message); + } + return; + } + + const data = await response.buffer(); + yield result.data(location, data); } catch (e) { - throw new Error(`Unable to read body of ${location.target}, ${e}`); + const message = `Unable to read ${location.type} ${location.target}, ${e}`; + yield result.generalError(location, message); } } diff --git a/plugins/catalog-backend/src/ingestion/processors/YamlProcessor.ts b/plugins/catalog-backend/src/ingestion/processors/YamlProcessor.ts index fabfb03939..291141c546 100644 --- a/plugins/catalog-backend/src/ingestion/processors/YamlProcessor.ts +++ b/plugins/catalog-backend/src/ingestion/processors/YamlProcessor.ts @@ -17,38 +17,39 @@ import { Entity, LocationSpec } from '@backstage/catalog-model'; import lodash from 'lodash'; import yaml from 'yaml'; -import { LocationProcessor, LocationProcessorResult } from './types'; +import { LocationProcessor, LocationProcessorResults } from './types'; +import * as result from './results'; export class YamlProcessor implements LocationProcessor { - async parseData( + async *parseData( data: Buffer, location: LocationSpec, - ): Promise { + ): LocationProcessorResults { if (!location.target.match(/\.ya?ml$/)) { - return undefined; + return; } let documents: yaml.Document.Parsed[]; try { documents = yaml.parseAllDocuments(data.toString('utf8')).filter(d => d); } catch (e) { - const error = new Error(`Failed to parse YAML, ${e}`); - return [{ type: 'error', location, error }]; + yield result.generalError(location, `Failed to parse YAML, ${e}`); + return; } - return documents.map(document => { + for (const document of documents) { if (document.errors?.length) { - const error = new Error(`YAML error, ${document.errors[0]}`); - return { type: 'error', location, error }; + const message = `YAML error, ${document.errors[0]}`; + yield result.generalError(location, message); + } else { + const json = document.toJSON(); + if (lodash.isPlainObject(json)) { + yield result.entity(location, json as Entity); + } else { + const message = `Expected object at root, got ${typeof json}`; + yield result.generalError(location, message); + } } - - const json = document.toJSON(); - if (lodash.isPlainObject(json)) { - return { type: 'entity', location, entity: json as Entity }; - } - - const error = new Error(`Expected object at root, got ${typeof json}`); - return { type: 'error', location, error }; - }); + } } } diff --git a/plugins/catalog-backend/src/ingestion/processors/results.ts b/plugins/catalog-backend/src/ingestion/processors/results.ts new file mode 100644 index 0000000000..9c6a7fc2b3 --- /dev/null +++ b/plugins/catalog-backend/src/ingestion/processors/results.ts @@ -0,0 +1,69 @@ +/* + * Copyright 2020 Spotify AB + * + * 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 { InputError, NotFoundError } from '@backstage/backend-common'; +import { Entity, LocationSpec } from '@backstage/catalog-model'; +import { LocationProcessorResult } from './types'; + +export function notFoundError( + atLocation: LocationSpec, + message: string, +): LocationProcessorResult { + return { + type: 'error', + location: atLocation, + error: new NotFoundError(message), + }; +} + +export function inputError( + atLocation: LocationSpec, + message: string, +): LocationProcessorResult { + return { + type: 'error', + location: atLocation, + error: new InputError(message), + }; +} + +export function generalError( + atLocation: LocationSpec, + message: string, +): LocationProcessorResult { + return { type: 'error', location: atLocation, error: new Error(message) }; +} + +export function data( + atLocation: LocationSpec, + newData: Buffer, +): LocationProcessorResult { + return { type: 'data', location: atLocation, data: newData }; +} + +export function location( + newLocation: LocationSpec, + optional: boolean, +): LocationProcessorResult { + return { type: 'location', location: newLocation, optional }; +} + +export function entity( + atLocation: LocationSpec, + newEntity: Entity, +): LocationProcessorResult { + return { type: 'entity', location: atLocation, entity: newEntity }; +} diff --git a/plugins/catalog-backend/src/ingestion/processors/types.ts b/plugins/catalog-backend/src/ingestion/processors/types.ts index 28840dd7df..0be7382c2a 100644 --- a/plugins/catalog-backend/src/ingestion/processors/types.ts +++ b/plugins/catalog-backend/src/ingestion/processors/types.ts @@ -21,27 +21,28 @@ export type LocationProcessor = { * Reads the contents of a location. * * @param location The location to read - * @returns The output if the location could be read successfully, or - * undefined if the location is not to be handled by this processor - * @throws NotFoundError if the location is handled by this reader, and the - * target did not exist - * @throws Any other Error if the location is handled by this reader, and it - * could not be read successfully */ readLocation?( location: LocationSpec, - ): Promise; + optional: boolean, + ): LocationProcessorResults; - parseData?( - data: Buffer, + parseData?(data: Buffer, location: LocationSpec): LocationProcessorResults; + + processEntity?( + entity: Entity, location: LocationSpec, - ): Promise; + ): LocationProcessorResults; - processEntity?(entity: Entity, location: LocationSpec): Promise; - - handleError?(error: Error, location: LocationSpec): Promise; + handleError?(error: Error, location: LocationSpec): LocationProcessorResults; }; +export type LocationProcessorResults = AsyncGenerator< + LocationProcessorResult, + void, + unknown +>; + export type LocationProcessorResult = | { type: 'error'; error: Error; location: LocationSpec } // An error occurred | { type: 'location'; location: LocationSpec; optional: boolean } // A location to read diff --git a/plugins/catalog-backend/src/service/standaloneServer.ts b/plugins/catalog-backend/src/service/standaloneServer.ts index 79cf72df4d..b4e1d5250e 100644 --- a/plugins/catalog-backend/src/service/standaloneServer.ts +++ b/plugins/catalog-backend/src/service/standaloneServer.ts @@ -16,13 +16,11 @@ import { Server } from 'http'; import { Logger } from 'winston'; -import { createStandaloneApplication } from './standaloneApplication'; import { DatabaseEntitiesCatalog } from '../catalog/DatabaseEntitiesCatalog'; -import { DatabaseManager } from '../database/DatabaseManager'; import { DatabaseLocationsCatalog } from '../catalog/DatabaseLocationsCatalog'; -import { LocationReaders } from '../ingestion/source/LocationReaders'; -import { IngestionModels, DescriptorParsers, HigherOrderOperations } from '..'; -import { EntityPolicies } from '@backstage/catalog-model'; +import { DatabaseManager } from '../database/DatabaseManager'; +import { HigherOrderOperations, LocationReaders } from '../ingestion'; +import { createStandaloneApplication } from './standaloneApplication'; export interface ServerOptions { port: number; @@ -39,15 +37,11 @@ export async function startStandaloneServer( const entitiesCatalog = new DatabaseEntitiesCatalog(db); const locationsCatalog = new DatabaseLocationsCatalog(db); - const ingestionModel = new IngestionModels( - new LocationReaders(), - new DescriptorParsers(), - new EntityPolicies(), - ); + const locationReader = new LocationReaders(); const higherOrderOperation = new HigherOrderOperations( entitiesCatalog, locationsCatalog, - ingestionModel, + locationReader, logger, );