From fe16d8ee8d8ba4756499a47fa4b24ca0ba4b4380 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Thu, 4 Jun 2020 13:57:38 +0200 Subject: [PATCH 1/5] chore(catalog-backend): make LocationReader, that runs a chain of processors --- packages/backend/src/plugins/catalog.ts | 13 +- packages/backend/tsconfig.json | 2 +- .../ingestion/HigherOrderOperations.test.ts | 66 +++--- .../src/ingestion/HigherOrderOperations.ts | 67 +++--- .../src/ingestion/IngestionModels.ts | 73 ------ .../src/ingestion/LocationReaders.ts | 215 ++++++++++++++++++ .../ingestion/descriptor/DescriptorParsers.ts | 45 ---- .../src/ingestion/descriptor/index.ts | 18 -- .../parsers/YamlDescriptorParser.ts | 64 ------ .../src/ingestion/descriptor/parsers/types.ts | 42 ---- .../catalog-backend/src/ingestion/index.ts | 13 +- .../AnnotateLocationEntityProcessor.ts} | 25 +- .../EntityPolicyProcessor.ts} | 18 +- .../processors/FileReaderProcessor.ts | 41 ++++ .../GithubReaderProcessor.ts} | 39 ++-- .../src/ingestion/processors/YamlProcessor.ts | 54 +++++ .../src/ingestion/processors/types.ts | 49 ++++ .../src/ingestion/source/LocationReaders.ts | 41 ---- .../readers/GitHubLocationReader.test.ts | 94 -------- .../src/ingestion/source/readers/types.ts | 29 --- .../catalog-backend/src/ingestion/types.ts | 41 +++- 21 files changed, 517 insertions(+), 532 deletions(-) delete mode 100644 plugins/catalog-backend/src/ingestion/IngestionModels.ts create mode 100644 plugins/catalog-backend/src/ingestion/LocationReaders.ts delete mode 100644 plugins/catalog-backend/src/ingestion/descriptor/DescriptorParsers.ts delete mode 100644 plugins/catalog-backend/src/ingestion/descriptor/index.ts delete mode 100644 plugins/catalog-backend/src/ingestion/descriptor/parsers/YamlDescriptorParser.ts delete mode 100644 plugins/catalog-backend/src/ingestion/descriptor/parsers/types.ts rename plugins/catalog-backend/src/ingestion/{source/readers/FileLocationReader.ts => processors/AnnotateLocationEntityProcessor.ts} (56%) rename plugins/catalog-backend/src/ingestion/{source/index.ts => processors/EntityPolicyProcessor.ts} (60%) create mode 100644 plugins/catalog-backend/src/ingestion/processors/FileReaderProcessor.ts rename plugins/catalog-backend/src/ingestion/{source/readers/GitHubLocationReader.ts => processors/GithubReaderProcessor.ts} (58%) create mode 100644 plugins/catalog-backend/src/ingestion/processors/YamlProcessor.ts create mode 100644 plugins/catalog-backend/src/ingestion/processors/types.ts delete mode 100644 plugins/catalog-backend/src/ingestion/source/LocationReaders.ts delete mode 100644 plugins/catalog-backend/src/ingestion/source/readers/GitHubLocationReader.test.ts delete mode 100644 plugins/catalog-backend/src/ingestion/source/readers/types.ts diff --git a/packages/backend/src/plugins/catalog.ts b/packages/backend/src/plugins/catalog.ts index 906e8c2d11..26f18d6b95 100644 --- a/packages/backend/src/plugins/catalog.ts +++ b/packages/backend/src/plugins/catalog.ts @@ -19,24 +19,17 @@ import { DatabaseEntitiesCatalog, DatabaseLocationsCatalog, DatabaseManager, - DescriptorParsers, - LocationReaders, - IngestionModels, - runPeriodically, HigherOrderOperations, + LocationReaders, + runPeriodically, } from '@backstage/plugin-catalog-backend'; import { PluginEnvironment } from '../types'; -import { EntityPolicies } from '@backstage/catalog-model'; export default async function createPlugin({ logger, database, }: PluginEnvironment) { - const ingestionModel = new IngestionModels( - new LocationReaders(), - new DescriptorParsers(), - new EntityPolicies(), - ); + const ingestionModel = new LocationReaders(); const db = await DatabaseManager.createDatabase(database, logger); const entitiesCatalog = new DatabaseEntitiesCatalog(db); diff --git a/packages/backend/tsconfig.json b/packages/backend/tsconfig.json index 015a967f76..04701a5502 100644 --- a/packages/backend/tsconfig.json +++ b/packages/backend/tsconfig.json @@ -9,7 +9,7 @@ "target": "es2019", "module": "commonjs", "esModuleInterop": true, - "lib": ["es2019"], + "lib": ["es2019", "dom"], "types": ["node", "jest"] } } diff --git a/plugins/catalog-backend/src/ingestion/HigherOrderOperations.test.ts b/plugins/catalog-backend/src/ingestion/HigherOrderOperations.test.ts index bd8dbedb27..b4dd151e72 100644 --- a/plugins/catalog-backend/src/ingestion/HigherOrderOperations.test.ts +++ b/plugins/catalog-backend/src/ingestion/HigherOrderOperations.test.ts @@ -15,17 +15,17 @@ */ import { getVoidLogger } from '@backstage/backend-common'; -import { Entity, Location } from '@backstage/catalog-model'; +import { Entity, Location, LocationSpec } from '@backstage/catalog-model'; import { EntitiesCatalog, LocationsCatalog } from '../catalog'; import { LocationUpdateStatus } from '../catalog/types'; import { DatabaseLocationUpdateLogStatus } from '../database/types'; import { HigherOrderOperations } from './HigherOrderOperations'; -import { IngestionModel } from './types'; +import { LocationReader } from './types'; describe('HigherOrderOperations', () => { let entitiesCatalog: jest.Mocked; let locationsCatalog: jest.Mocked; - let ingestionModel: jest.Mocked; + let locationReader: jest.Mocked; let higherOrderOperation: HigherOrderOperations; beforeAll(() => { @@ -45,13 +45,13 @@ describe('HigherOrderOperations', () => { logUpdateSuccess: jest.fn(), logUpdateFailure: jest.fn(), }; - ingestionModel = { - readLocation: jest.fn(), + locationReader = { + read: jest.fn(), }; higherOrderOperation = new HigherOrderOperations( entitiesCatalog, locationsCatalog, - ingestionModel, + locationReader, getVoidLogger(), ); }); @@ -68,7 +68,7 @@ describe('HigherOrderOperations', () => { }; locationsCatalog.addLocation.mockImplementation(x => Promise.resolve(x)); locationsCatalog.locations.mockResolvedValue([]); - ingestionModel.readLocation.mockResolvedValue([]); + locationReader.read.mockResolvedValue({ entities: [], errors: [] }); const result = await higherOrderOperation.addLocation(spec); @@ -80,8 +80,8 @@ describe('HigherOrderOperations', () => { ); expect(result.entities).toEqual([]); expect(locationsCatalog.locations).toBeCalledTimes(1); - expect(ingestionModel.readLocation).toBeCalledTimes(1); - expect(ingestionModel.readLocation).toBeCalledWith('a', 'b'); + expect(locationReader.read).toBeCalledTimes(1); + expect(locationReader.read).toBeCalledWith({ type: 'a', target: 'b' }); expect(entitiesCatalog.addOrUpdateEntity).not.toBeCalled(); expect(locationsCatalog.addLocation).toBeCalledTimes(1); expect(locationsCatalog.addLocation).toBeCalledWith( @@ -108,15 +108,15 @@ describe('HigherOrderOperations', () => { data: location, }, ]); - ingestionModel.readLocation.mockResolvedValue([]); + locationReader.read.mockResolvedValue({ entities: [], errors: [] }); const result = await higherOrderOperation.addLocation(spec); expect(result.location).toEqual(location); expect(result.entities).toEqual([]); expect(locationsCatalog.locations).toBeCalledTimes(1); - expect(ingestionModel.readLocation).toBeCalledTimes(1); - expect(ingestionModel.readLocation).toBeCalledWith('a', 'b'); + expect(locationReader.read).toBeCalledTimes(1); + expect(locationReader.read).toBeCalledWith({ type: 'a', target: 'b' }); expect(entitiesCatalog.addOrUpdateEntity).not.toBeCalled(); expect(locationsCatalog.addLocation).not.toBeCalled(); }); @@ -126,6 +126,7 @@ describe('HigherOrderOperations', () => { type: 'a', target: 'b', }; + const location: LocationSpec = { type: '', target: '' }; const entity: Entity = { apiVersion: 'a', kind: 'b', @@ -133,10 +134,10 @@ describe('HigherOrderOperations', () => { }; locationsCatalog.locations.mockResolvedValue([]); - ingestionModel.readLocation.mockResolvedValue([ - { type: 'data', data: entity }, - { type: 'error', error: new Error('abcd') }, - ]); + locationReader.read.mockResolvedValue({ + entities: [{ entity, location }], + errors: [{ error: new Error('abcd'), location }], + }); await expect(higherOrderOperation.addLocation(spec)).rejects.toThrow( /abcd/, @@ -156,7 +157,7 @@ describe('HigherOrderOperations', () => { ).resolves.toBeUndefined(); expect(locationsCatalog.locations).toHaveBeenCalledTimes(1); - expect(ingestionModel.readLocation).not.toHaveBeenCalled(); + expect(locationReader.read).not.toHaveBeenCalled(); expect(entitiesCatalog.addOrUpdateEntity).not.toHaveBeenCalled(); }); @@ -181,9 +182,10 @@ describe('HigherOrderOperations', () => { locationsCatalog.locations.mockResolvedValue([ { currentStatus: locationStatus, data: location }, ]); - ingestionModel.readLocation.mockResolvedValue([ - { type: 'data', data: desc }, - ]); + locationReader.read.mockResolvedValue({ + entities: [{ entity: desc, location }], + errors: [], + }); entitiesCatalog.entityByName.mockResolvedValue(undefined); entitiesCatalog.addOrUpdateEntity.mockResolvedValue(desc); @@ -192,12 +194,11 @@ describe('HigherOrderOperations', () => { ).resolves.toBeUndefined(); expect(locationsCatalog.locations).toHaveBeenCalledTimes(1); - expect(ingestionModel.readLocation).toHaveBeenCalledTimes(1); - expect(ingestionModel.readLocation).toHaveBeenNthCalledWith( - 1, - 'some', - 'thing', - ); + expect(locationReader.read).toHaveBeenCalledTimes(1); + expect(locationReader.read).toHaveBeenNthCalledWith(1, { + type: 'some', + target: 'thing', + }); expect(entitiesCatalog.entityByName).toHaveBeenCalledTimes(1); expect(entitiesCatalog.entityByName).toHaveBeenNthCalledWith( 1, @@ -236,9 +237,10 @@ describe('HigherOrderOperations', () => { locationsCatalog.locations.mockResolvedValue([ { currentStatus: locationStatus, data: location }, ]); - ingestionModel.readLocation.mockResolvedValue([ - { type: 'data', data: desc }, - ]); + locationReader.read.mockResolvedValue({ + entities: [{ entity: desc, location }], + errors: [], + }); entitiesCatalog.entityByName.mockResolvedValue(undefined); entitiesCatalog.addOrUpdateEntity.mockResolvedValue(desc); @@ -272,15 +274,13 @@ describe('HigherOrderOperations', () => { locationsCatalog.locations.mockResolvedValue([ { currentStatus: locationStatus, data: location }, ]); - ingestionModel.readLocation.mockRejectedValue( - new Error('reader error message'), - ); + locationReader.read.mockRejectedValue(new Error('reader error message')); await expect( higherOrderOperation.refreshAllLocations(), ).resolves.toBeUndefined(); - expect(ingestionModel.readLocation).toHaveBeenCalledTimes(1); + expect(locationReader.read).toHaveBeenCalledTimes(1); expect(locationsCatalog.logUpdateFailure).toHaveBeenCalledTimes(1); expect(locationsCatalog.logUpdateSuccess).not.toHaveBeenCalled(); expect(locationsCatalog.logUpdateFailure).toHaveBeenCalledWith( diff --git a/plugins/catalog-backend/src/ingestion/HigherOrderOperations.ts b/plugins/catalog-backend/src/ingestion/HigherOrderOperations.ts index 3300df1640..bc346f8a4d 100644 --- a/plugins/catalog-backend/src/ingestion/HigherOrderOperations.ts +++ b/plugins/catalog-backend/src/ingestion/HigherOrderOperations.ts @@ -24,8 +24,11 @@ import { import lodash from 'lodash'; import { v4 as uuidv4 } from 'uuid'; import { EntitiesCatalog, LocationsCatalog } from '../catalog'; -import { IngestionModel } from '../ingestion'; -import { AddLocationResult, HigherOrderOperation } from './types'; +import { + AddLocationResult, + HigherOrderOperation, + LocationReader, +} from './types'; import { Logger } from 'winston'; /** @@ -38,18 +41,18 @@ import { Logger } from 'winston'; export class HigherOrderOperations implements HigherOrderOperation { private readonly entitiesCatalog: EntitiesCatalog; private readonly locationsCatalog: LocationsCatalog; - private readonly ingestionModel: IngestionModel; + private readonly locationReader: LocationReader; private readonly logger: Logger; constructor( entitiesCatalog: EntitiesCatalog, locationsCatalog: LocationsCatalog, - ingestionModel: IngestionModel, + locationReader: LocationReader, logger: Logger, ) { this.entitiesCatalog = entitiesCatalog; this.locationsCatalog = locationsCatalog; - this.ingestionModel = ingestionModel; + this.locationReader = locationReader; this.logger = logger; } @@ -80,28 +83,16 @@ export class HigherOrderOperations implements HigherOrderOperation { }; // Read the location fully, bailing on any errors - const readerOutput = await this.ingestionModel.readLocation( - location.type, - location.target, - ); - const inputEntities: Entity[] = []; - for (const entry of readerOutput) { - if (entry.type === 'error') { - throw new InputError( - `Failed to read location ${location.type} ${location.target}, ${entry.error}`, - ); - } else { - // Append the location reference annotation - entry.data.metadata.annotations = { - ...entry.data.metadata.annotations, - [LOCATION_ANNOTATION]: location.id, - }; - inputEntities.push(entry.data); - } + const readerOutput = await this.locationReader.read(spec); + if (readerOutput.errors.length) { + const item = readerOutput.errors[0]; + throw new InputError( + `Failed to read location ${item.location.type} ${item.location.target}, ${item.error}`, + ); } // TODO(freben): At this point, we could detect orphaned entities, by way - // of having a LOCATION_ANNOTATION pointing to the location but not being + // of having a location annotation pointing to the location but not being // in the entities list. But we aren't sure what to do about those yet. // Write @@ -109,9 +100,9 @@ export class HigherOrderOperations implements HigherOrderOperation { await this.locationsCatalog.addLocation(location); } const outputEntities: Entity[] = []; - for (const entity of inputEntities) { + for (const entity of readerOutput.entities) { const out = await this.entitiesCatalog.addOrUpdateEntity( - entity, + entity.entity, location.id, ); outputEntities.push(out); @@ -157,20 +148,20 @@ export class HigherOrderOperations implements HigherOrderOperation { // Performs a full refresh of a single location private async refreshSingleLocation(location: Location) { - const readerOutput = await this.ingestionModel.readLocation( - location.type, - location.target, - ); + const readerOutput = await this.locationReader.read({ + type: location.type, + target: location.target, + }); - for (const readerItem of readerOutput) { - if (readerItem.type === 'error') { - this.logger.debug( - `Failed item in location id="${location.id}" type="${location.type}" target="${location.target}", ${readerItem.error}`, - ); - continue; - } + for (const item of readerOutput.errors) { + this.logger.debug( + `Failed item in location type="${item.location.type}" target="${item.location.target}", ${item.error}`, + ); + } + + for (const item of readerOutput.entities) { + const { entity } = item; - const entity = readerItem.data; this.logger.debug( `Read entity kind="${entity.kind}" name="${ entity.metadata.name diff --git a/plugins/catalog-backend/src/ingestion/IngestionModels.ts b/plugins/catalog-backend/src/ingestion/IngestionModels.ts deleted file mode 100644 index 8febdecd18..0000000000 --- a/plugins/catalog-backend/src/ingestion/IngestionModels.ts +++ /dev/null @@ -1,73 +0,0 @@ -/* - * 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 { EntityPolicies, EntityPolicy } from '@backstage/catalog-model'; -import { DescriptorParsers } from './descriptor'; -import { DescriptorParser, ReaderOutput } from './descriptor/parsers/types'; -import { LocationReader, LocationReaders } from './source'; -import { IngestionModel } from './types'; - -export class IngestionModels implements IngestionModel { - private readonly reader: LocationReader; - private readonly parser: DescriptorParser; - private readonly entityPolicy: EntityPolicy; - - static default(): IngestionModel { - return new IngestionModels( - new LocationReaders(), - new DescriptorParsers(), - new EntityPolicies(), - ); - } - - constructor( - reader: LocationReader, - parser: DescriptorParser, - entityPolicy: EntityPolicy, - ) { - this.reader = reader; - this.parser = parser; - this.entityPolicy = entityPolicy; - } - - async readLocation(type: string, target: string): Promise { - const buffer = await this.reader.tryRead(type, target); - if (!buffer) { - throw new Error(`No reader could handle location ${type} ${target}`); - } - - const items = await this.parser.tryParse(buffer); - if (!items) { - throw new Error(`No parser could handle location ${type} ${target}`); - } - - const result: ReaderOutput[] = []; - for (const item of items) { - if (item.type === 'error') { - result.push(item); - } else { - try { - const output = await this.entityPolicy.enforce(item.data); - result.push({ type: 'data', data: output }); - } catch (e) { - result.push({ type: 'error', error: e }); - } - } - } - - return result; - } -} diff --git a/plugins/catalog-backend/src/ingestion/LocationReaders.ts b/plugins/catalog-backend/src/ingestion/LocationReaders.ts new file mode 100644 index 0000000000..92c11d1138 --- /dev/null +++ b/plugins/catalog-backend/src/ingestion/LocationReaders.ts @@ -0,0 +1,215 @@ +/* + * 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 { NotFoundError } from '@backstage/backend-common'; +import { + Entity, + EntityPolicies, + EntityPolicy, + LocationSpec, +} from '@backstage/catalog-model'; +import { AnnotateLocationEntityProcessor } from './processors/AnnotateLocationEntityProcessor'; +import { EntityPolicyProcessor } from './processors/EntityPolicyProcessor'; +import { FileReaderProcessor } from './processors/FileReaderProcessor'; +import { GithubReaderProcessor } from './processors/GithubReaderProcessor'; +import { LocationProcessor, LocationProcessorResult } 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 }; + +/** + * Implements the reading of a location through a series of processor tasks. + */ +export class LocationReaders implements LocationReader { + private readonly processors: LocationProcessor[]; + + static defaultProcessors( + entityPolicy: EntityPolicy = new EntityPolicies(), + ): LocationProcessor[] { + return [ + new FileReaderProcessor(), + new GithubReaderProcessor(), + new YamlProcessor(), + new EntityPolicyProcessor(entityPolicy), + new AnnotateLocationEntityProcessor(), + ]; + } + + constructor( + processors: LocationProcessor[] = LocationReaders.defaultProcessors(), + ) { + this.processors = processors; + } + + 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', + location, + depth, + error: new Error( + `No processor could read location ${location.type} ${location.target}`, + ), + }); + } + + async handleData( + data: Buffer, + location: LocationSpec, + depth: number, + queue: QueueItem[], + ): 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; + } + } + + 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, + }); + } + } +} diff --git a/plugins/catalog-backend/src/ingestion/descriptor/DescriptorParsers.ts b/plugins/catalog-backend/src/ingestion/descriptor/DescriptorParsers.ts deleted file mode 100644 index ed05855109..0000000000 --- a/plugins/catalog-backend/src/ingestion/descriptor/DescriptorParsers.ts +++ /dev/null @@ -1,45 +0,0 @@ -/* - * 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 { DescriptorParser, ReaderOutput } from './parsers/types'; -import { YamlDescriptorParser } from './parsers/YamlDescriptorParser'; - -/** - * Parses raw descriptor data (e.g. from a file or stream) into entities. - */ -export class DescriptorParsers implements DescriptorParser { - private readonly parsers: DescriptorParser[]; - - static defaultParsers(): DescriptorParser[] { - return [new YamlDescriptorParser()]; - } - - constructor( - parsers: DescriptorParser[] = DescriptorParsers.defaultParsers(), - ) { - this.parsers = parsers; - } - - async tryParse(data: Buffer): Promise { - for (const parser of this.parsers) { - const result = await parser.tryParse(data); - if (result) { - return result; - } - } - throw new Error(`Unsupported descriptor format`); - } -} diff --git a/plugins/catalog-backend/src/ingestion/descriptor/index.ts b/plugins/catalog-backend/src/ingestion/descriptor/index.ts deleted file mode 100644 index 1529c78afc..0000000000 --- a/plugins/catalog-backend/src/ingestion/descriptor/index.ts +++ /dev/null @@ -1,18 +0,0 @@ -/* - * 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. - */ - -export { DescriptorParsers } from './DescriptorParsers'; -export { YamlDescriptorParser } from './parsers/YamlDescriptorParser'; diff --git a/plugins/catalog-backend/src/ingestion/descriptor/parsers/YamlDescriptorParser.ts b/plugins/catalog-backend/src/ingestion/descriptor/parsers/YamlDescriptorParser.ts deleted file mode 100644 index 3f2e9e3c5c..0000000000 --- a/plugins/catalog-backend/src/ingestion/descriptor/parsers/YamlDescriptorParser.ts +++ /dev/null @@ -1,64 +0,0 @@ -/* - * 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 { Entity } from '@backstage/catalog-model'; -import yaml from 'yaml'; -import { DescriptorParser, ReaderOutput } from './types'; - -/** - * Parses descriptors on YAML format - */ -export class YamlDescriptorParser implements DescriptorParser { - async tryParse(data: Buffer): Promise { - // TODO(freben): Should perhaps first do format detection, so the parse - // failure can be emitted as a proper error instead of just as if we - // weren't handling the format at all. - let documents; - try { - documents = yaml.parseAllDocuments(data.toString('utf8')); - } catch (e) { - return undefined; - } - - const result: ReaderOutput[] = []; - - for (const document of documents) { - if (document.contents) { - if (document.errors?.length) { - result.push({ - type: 'error', - error: new Error(`Malformed YAML document, ${document.errors[0]}`), - }); - } else { - const json = document.toJSON(); - if (typeof json !== 'object' || Array.isArray(json)) { - result.push({ - type: 'error', - error: new Error(`Malformed descriptor, expected object at root`), - }); - } else { - result.push({ - type: 'data', - data: json as Entity, - }); - } - } - } - } - - return result; - } -} diff --git a/plugins/catalog-backend/src/ingestion/descriptor/parsers/types.ts b/plugins/catalog-backend/src/ingestion/descriptor/parsers/types.ts deleted file mode 100644 index baef5dc3df..0000000000 --- a/plugins/catalog-backend/src/ingestion/descriptor/parsers/types.ts +++ /dev/null @@ -1,42 +0,0 @@ -/* - * 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 { Entity } from '@backstage/catalog-model'; - -export type ReaderOutput = - | { type: 'error'; error: Error } - | { type: 'data'; data: Entity }; - -/** - * Parses raw descriptor data (e.g. from a file) into entities. - */ -export type DescriptorParser = { - /** - * Try to parse some raw data into an entity. - * - * Note that this is only the low level operation of parsing the raw file - * format, e.g. reading JSON or YAML or similar and emitting as structured - * but unvalidated data. The actual validation is performed by EntityPolicy - * and KindParser. - * - * @param data Raw descriptor data - * @returns A list of raw unvalidated entities / errors, or undefined if the - * given data is not meant to be handled by this parser - * @throws An Error if the format was handled and found to not be properly - * formed - */ - tryParse(data: Buffer): Promise; -}; diff --git a/plugins/catalog-backend/src/ingestion/index.ts b/plugins/catalog-backend/src/ingestion/index.ts index 6c530c7234..dc9bce56f2 100644 --- a/plugins/catalog-backend/src/ingestion/index.ts +++ b/plugins/catalog-backend/src/ingestion/index.ts @@ -14,8 +14,13 @@ * limitations under the License. */ -export * from './descriptor'; export { HigherOrderOperations } from './HigherOrderOperations'; -export { IngestionModels } from './IngestionModels'; -export * from './source'; -export type { HigherOrderOperation, IngestionModel } from './types'; +export { LocationReaders } from './LocationReaders'; +export type { + HigherOrderOperation, + AddLocationResult, + LocationReader, + ReadLocationResult, + ReadLocationEntity, + ReadLocationError, +} from './types'; diff --git a/plugins/catalog-backend/src/ingestion/source/readers/FileLocationReader.ts b/plugins/catalog-backend/src/ingestion/processors/AnnotateLocationEntityProcessor.ts similarity index 56% rename from plugins/catalog-backend/src/ingestion/source/readers/FileLocationReader.ts rename to plugins/catalog-backend/src/ingestion/processors/AnnotateLocationEntityProcessor.ts index 0c64aebf14..fac6298660 100644 --- a/plugins/catalog-backend/src/ingestion/source/readers/FileLocationReader.ts +++ b/plugins/catalog-backend/src/ingestion/processors/AnnotateLocationEntityProcessor.ts @@ -14,22 +14,15 @@ * limitations under the License. */ -import fs from 'fs-extra'; -import { LocationReader } from './types'; +import { Entity, LocationSpec } from '@backstage/catalog-model'; +import lodash from 'lodash'; +import { LocationProcessor } from './types'; -/** - * Reads a file from the local file system. - */ -export class FileLocationReader implements LocationReader { - async tryRead(type: string, target: string): Promise { - if (type !== 'file') { - return undefined; - } - - try { - return await fs.readFile(target); - } catch (e) { - throw new Error(`Unable to read "${target}", ${e}`); - } +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); } } diff --git a/plugins/catalog-backend/src/ingestion/source/index.ts b/plugins/catalog-backend/src/ingestion/processors/EntityPolicyProcessor.ts similarity index 60% rename from plugins/catalog-backend/src/ingestion/source/index.ts rename to plugins/catalog-backend/src/ingestion/processors/EntityPolicyProcessor.ts index db7aa2f0bd..fcc7c71578 100644 --- a/plugins/catalog-backend/src/ingestion/source/index.ts +++ b/plugins/catalog-backend/src/ingestion/processors/EntityPolicyProcessor.ts @@ -14,7 +14,17 @@ * limitations under the License. */ -export { LocationReaders } from './LocationReaders'; -export { FileLocationReader } from './readers/FileLocationReader'; -export { GitHubLocationReader } from './readers/GitHubLocationReader'; -export type { LocationReader } from './readers/types'; +import { Entity, EntityPolicy } from '@backstage/catalog-model'; +import { LocationProcessor } from './types'; + +export class EntityPolicyProcessor implements LocationProcessor { + private readonly policy: EntityPolicy; + + constructor(policy: EntityPolicy) { + this.policy = policy; + } + + async processEntity(entity: Entity): Promise { + return this.policy.enforce(entity); + } +} diff --git a/plugins/catalog-backend/src/ingestion/processors/FileReaderProcessor.ts b/plugins/catalog-backend/src/ingestion/processors/FileReaderProcessor.ts new file mode 100644 index 0000000000..6aa9100506 --- /dev/null +++ b/plugins/catalog-backend/src/ingestion/processors/FileReaderProcessor.ts @@ -0,0 +1,41 @@ +/* + * 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 { NotFoundError } from '@backstage/backend-common'; +import { LocationSpec } from '@backstage/catalog-model'; +import fs from 'fs-extra'; +import { LocationProcessor, LocationProcessorResult } from './types'; + +export class FileReaderProcessor implements LocationProcessor { + async readLocation( + location: LocationSpec, + ): Promise { + if (location.type !== 'file') { + return undefined; + } + + if (!(await fs.pathExists(location.target))) { + throw new NotFoundError(`${location.target} does not exist`); + } + + try { + const data = await fs.readFile(location.target); + return [{ type: 'data', location, data }]; + } catch (e) { + throw new Error(`Unable to read ${location.target}, ${e}`); + } + } +} diff --git a/plugins/catalog-backend/src/ingestion/source/readers/GitHubLocationReader.ts b/plugins/catalog-backend/src/ingestion/processors/GithubReaderProcessor.ts similarity index 58% rename from plugins/catalog-backend/src/ingestion/source/readers/GitHubLocationReader.ts rename to plugins/catalog-backend/src/ingestion/processors/GithubReaderProcessor.ts index bd330e28b4..1533f0a633 100644 --- a/plugins/catalog-backend/src/ingestion/source/readers/GitHubLocationReader.ts +++ b/plugins/catalog-backend/src/ingestion/processors/GithubReaderProcessor.ts @@ -14,30 +14,41 @@ * limitations under the License. */ +import { NotFoundError } from '@backstage/backend-common'; +import { LocationSpec } from '@backstage/catalog-model'; import fetch from 'node-fetch'; -import { URL } from 'url'; -import { LocationReader } from './types'; +import { LocationProcessor, LocationProcessorResult } from './types'; -/** - * Reads a file whose target is a GitHub URL. - * - * Uses raw.githubusercontent.com for now, but this will probably change in the - * future when token auth is implemented. - */ -export class GitHubLocationReader implements LocationReader { - async tryRead(type: string, target: string): Promise { - if (type !== 'github') { +export class GithubReaderProcessor implements LocationProcessor { + async readLocation( + location: LocationSpec, + ): Promise { + if (location.type !== 'github') { return undefined; } - const url = this.buildRawUrl(target); + 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); + } + } + try { - return await fetch(url.toString()).then(x => x.buffer()); + return [{ type: 'data', location, data: await response.buffer() }]; } catch (e) { - throw new Error(`Unable to read "${target}", ${e}`); + throw new Error(`Unable to read body of ${location.target}, ${e}`); } } + // Converts + // from: https://github.com/a/b/blob/master/c.yaml + // to: https://raw.githubusercontent.com/a/b/master/c.yaml private buildRawUrl(target: string): URL { try { const url = new URL(target); diff --git a/plugins/catalog-backend/src/ingestion/processors/YamlProcessor.ts b/plugins/catalog-backend/src/ingestion/processors/YamlProcessor.ts new file mode 100644 index 0000000000..fabfb03939 --- /dev/null +++ b/plugins/catalog-backend/src/ingestion/processors/YamlProcessor.ts @@ -0,0 +1,54 @@ +/* + * 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 { Entity, LocationSpec } from '@backstage/catalog-model'; +import lodash from 'lodash'; +import yaml from 'yaml'; +import { LocationProcessor, LocationProcessorResult } from './types'; + +export class YamlProcessor implements LocationProcessor { + async parseData( + data: Buffer, + location: LocationSpec, + ): Promise { + if (!location.target.match(/\.ya?ml$/)) { + return undefined; + } + + 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 }]; + } + + return documents.map(document => { + if (document.errors?.length) { + const error = new Error(`YAML error, ${document.errors[0]}`); + return { type: 'error', location, error }; + } + + 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/types.ts b/plugins/catalog-backend/src/ingestion/processors/types.ts new file mode 100644 index 0000000000..28840dd7df --- /dev/null +++ b/plugins/catalog-backend/src/ingestion/processors/types.ts @@ -0,0 +1,49 @@ +/* + * 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 { LocationSpec, Entity } from '@backstage/catalog-model'; + +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; + + parseData?( + data: Buffer, + location: LocationSpec, + ): Promise; + + processEntity?(entity: Entity, location: LocationSpec): Promise; + + handleError?(error: Error, location: LocationSpec): Promise; +}; + +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 diff --git a/plugins/catalog-backend/src/ingestion/source/LocationReaders.ts b/plugins/catalog-backend/src/ingestion/source/LocationReaders.ts deleted file mode 100644 index a670f309ad..0000000000 --- a/plugins/catalog-backend/src/ingestion/source/LocationReaders.ts +++ /dev/null @@ -1,41 +0,0 @@ -/* - * 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 { FileLocationReader } from './readers/FileLocationReader'; -import { GitHubLocationReader } from './readers/GitHubLocationReader'; -import { LocationReader } from './readers/types'; - -export class LocationReaders implements LocationReader { - private readonly readers: LocationReader[]; - - static defaultReaders(): LocationReader[] { - return [new FileLocationReader(), new GitHubLocationReader()]; - } - - constructor(readers: LocationReader[] = LocationReaders.defaultReaders()) { - this.readers = readers; - } - - async tryRead(type: string, target: string): Promise { - for (const reader of this.readers) { - const result = await reader.tryRead(type, target); - if (result) { - return result; - } - } - throw new Error(`Could not read unknown location "${type}", "${target}"`); - } -} diff --git a/plugins/catalog-backend/src/ingestion/source/readers/GitHubLocationReader.test.ts b/plugins/catalog-backend/src/ingestion/source/readers/GitHubLocationReader.test.ts deleted file mode 100644 index 56120fab23..0000000000 --- a/plugins/catalog-backend/src/ingestion/source/readers/GitHubLocationReader.test.ts +++ /dev/null @@ -1,94 +0,0 @@ -/* - * 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. - */ - -jest.mock('node-fetch'); - -import fetch from 'node-fetch'; -import { GitHubLocationReader } from './GitHubLocationReader'; - -const { Response } = jest.requireActual('node-fetch'); - -describe('Unit: GitHubLocationReader', () => { - beforeEach(() => { - jest.clearAllMocks(); - }); - - it('fetches the file and parses it correctly', async () => { - (fetch as any).mockResolvedValueOnce(new Response('hello')); - - const reader = new GitHubLocationReader(); - const buffer = await reader.tryRead( - 'github', - 'https://github.com/spotify/backstage/blob/master/plugins/catalog-backend/fixtures/one_component.yaml', - ); - - expect(buffer?.toString('utf8')).toBe('hello'); - }); - - it('changes the url to point to https://raw.githubusercontent.com', async () => { - const gitHubUrl = `https://github.com`; - const project = `spotify/backstage`; - const folderPath = `master/plugins/catalog-backend/fixtures`; - const componentFilename = `one_component.yaml`; - const rawGitHubUrl = `https://raw.githubusercontent.com`; - - const reader = new GitHubLocationReader(); - (fetch as any).mockResolvedValueOnce(new Response('hello')); - - await reader.tryRead( - 'github', - `${gitHubUrl}/${project}/blob/${folderPath}/${componentFilename}`, - ); - - expect(fetch).toHaveBeenCalledWith( - `${rawGitHubUrl}/${project}/${folderPath}/${componentFilename}`, - ); - }); - - describe('rejects wrong urls', () => { - const reader = new GitHubLocationReader(); - - it.each([ - ['http://example.com/one_component.yaml'], - ['http://github.com/one_component.yaml'], - ['http://github.com/PROJECT/one_component.yaml'], - ['http://github.com/PROJECT/REPO/one_component.yaml'], - ['http://github.com/PROJECT/REPO/one_component.json'], - ])( - '%p', - async (url: string) => - await expect(reader.tryRead('github', url)).rejects.toThrow(/url/), - ); - }); -}); - -describe('Integration: GitHubLocationSource', () => { - beforeAll(() => { - (fetch as any).mockImplementation(jest.requireActual('node-fetch')); - }); - - it('fetches the fixture from backstage repo', async () => { - (fetch as any).mockResolvedValueOnce(new Response('component3')); - - const PERMANENT_LINK = - 'https://github.com/spotify/backstage/blob/ee84a874f8e37f87940cbe515a86c07a2db29541/plugins/catalog-backend/fixtures/one_component.yaml'; - - const reader = new GitHubLocationReader(); - const result = await reader.tryRead('github', PERMANENT_LINK); - - expect(result?.toString('utf8')).toContain('component3'); - }); -}); diff --git a/plugins/catalog-backend/src/ingestion/source/readers/types.ts b/plugins/catalog-backend/src/ingestion/source/readers/types.ts deleted file mode 100644 index 37c7b46885..0000000000 --- a/plugins/catalog-backend/src/ingestion/source/readers/types.ts +++ /dev/null @@ -1,29 +0,0 @@ -/* - * 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. - */ - -export type LocationReader = { - /** - * Reads the contents of a single location. - * - * @param type The type of location to read - * @param target The location target (type-specific) - * @returns The target contents, as a raw Buffer, or undefined if this type - * was not meant to be consumed by this reader - * @throws An error if the type was meant for this reader, but could not be - * read - */ - tryRead(type: string, target: string): Promise; -}; diff --git a/plugins/catalog-backend/src/ingestion/types.ts b/plugins/catalog-backend/src/ingestion/types.ts index 8fb56367a6..5bbec01e7e 100644 --- a/plugins/catalog-backend/src/ingestion/types.ts +++ b/plugins/catalog-backend/src/ingestion/types.ts @@ -15,18 +15,47 @@ */ import type { Entity, Location, LocationSpec } from '@backstage/catalog-model'; -import type { ReaderOutput } from './descriptor/parsers/types'; + +// +// HigherOrderOperation +// + +export type HigherOrderOperation = { + addLocation(spec: LocationSpec): Promise; + refreshAllLocations(): Promise; +}; export type AddLocationResult = { location: Location; entities: Entity[]; }; -export type IngestionModel = { - readLocation(type: string, target: string): Promise; +// +// LocationReader +// + +export type LocationReader = { + /** + * Reads the contents of a location. + * + * @param location The location to read + * @throws An error if the location was handled by this reader, but could not + * be read + */ + read(location: LocationSpec): Promise; }; -export type HigherOrderOperation = { - addLocation(spec: LocationSpec): Promise; - refreshAllLocations(): Promise; +export type ReadLocationResult = { + entities: ReadLocationEntity[]; + errors: ReadLocationError[]; +}; + +export type ReadLocationEntity = { + location: LocationSpec; + entity: Entity; +}; + +export type ReadLocationError = { + location: LocationSpec; + error: Error; }; 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 2/5] 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, ); From 61a02be0f4d151f6eacb12516a604f1f4d4d4537 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Fri, 5 Jun 2020 14:50:19 +0200 Subject: [PATCH 3/5] Add logging and fix last bug --- packages/backend/src/plugins/catalog.ts | 2 +- .../src/ingestion/LocationReaders.ts | 39 ++++++++++++++++--- .../src/service/standaloneServer.ts | 2 +- 3 files changed, 35 insertions(+), 8 deletions(-) diff --git a/packages/backend/src/plugins/catalog.ts b/packages/backend/src/plugins/catalog.ts index ccd5d06908..8d8ac5b47b 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 locationReader = new LocationReaders(); + const locationReader = new LocationReaders(logger); const db = await DatabaseManager.createDatabase(database, logger); const entitiesCatalog = new DatabaseEntitiesCatalog(db); diff --git a/plugins/catalog-backend/src/ingestion/LocationReaders.ts b/plugins/catalog-backend/src/ingestion/LocationReaders.ts index 92f1a235ee..a40e86d177 100644 --- a/plugins/catalog-backend/src/ingestion/LocationReaders.ts +++ b/plugins/catalog-backend/src/ingestion/LocationReaders.ts @@ -14,11 +14,13 @@ * limitations under the License. */ +import { getVoidLogger } from '@backstage/backend-common'; import { EntityPolicies, EntityPolicy, LocationSpec, } from '@backstage/catalog-model'; +import { Logger } from 'winston'; import { AnnotateLocationEntityProcessor } from './processors/AnnotateLocationEntityProcessor'; import { EntityPolicyProcessor } from './processors/EntityPolicyProcessor'; import { FileReaderProcessor } from './processors/FileReaderProcessor'; @@ -38,6 +40,7 @@ const MAX_DEPTH = 10; * Implements the reading of a location through a series of processor tasks. */ export class LocationReaders implements LocationReader { + private readonly logger: Logger; private readonly processors: LocationProcessor[]; static defaultProcessors( @@ -53,8 +56,10 @@ export class LocationReaders implements LocationReader { } constructor( + logger: Logger = getVoidLogger(), processors: LocationProcessor[] = LocationReaders.defaultProcessors(), ) { + this.logger = logger; this.processors = processors; } @@ -76,15 +81,21 @@ export class LocationReaders implements LocationReader { ): 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(`Max recursion depth ${MAX_DEPTH} reached`), + error: new Error(message), }); return; } 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, @@ -92,7 +103,11 @@ export class LocationReaders implements LocationReader { 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, @@ -100,7 +115,11 @@ export class LocationReaders implements LocationReader { 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, @@ -112,8 +131,12 @@ export class LocationReaders implements LocationReader { 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 => { @@ -136,6 +159,7 @@ export class LocationReaders implements LocationReader { } async runAll( + what: string, start: ( processor: LocationProcessor, ) => LocationProcessorResults | undefined, @@ -157,6 +181,9 @@ export class LocationReaders implements LocationReader { if (item.done) { break; } + if (!item.value) { + continue; + } wasHandled = true; await emit(item.value); @@ -166,15 +193,15 @@ export class LocationReaders implements LocationReader { return; } } catch (e) { - const message = `Processor ${processor.constructor.name} threw an error, ${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}`; - await emit({ type: 'error', location, error: new Error(message) }); - } + 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) }); } } } diff --git a/plugins/catalog-backend/src/service/standaloneServer.ts b/plugins/catalog-backend/src/service/standaloneServer.ts index b4e1d5250e..11fbd0b445 100644 --- a/plugins/catalog-backend/src/service/standaloneServer.ts +++ b/plugins/catalog-backend/src/service/standaloneServer.ts @@ -37,7 +37,7 @@ export async function startStandaloneServer( const entitiesCatalog = new DatabaseEntitiesCatalog(db); const locationsCatalog = new DatabaseLocationsCatalog(db); - const locationReader = new LocationReaders(); + const locationReader = new LocationReaders(options.logger); const higherOrderOperation = new HigherOrderOperations( entitiesCatalog, locationsCatalog, 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 4/5] 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; From 9f02cf4a3027984b5151fb1ea57054b100c4c2ad Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Mon, 8 Jun 2020 07:14:16 +0200 Subject: [PATCH 5/5] LocationProcessorSink -> LocationProcessorEmit --- .../src/ingestion/LocationReaders.ts | 40 ++++++++++--------- .../catalog-backend/src/ingestion/index.ts | 4 +- .../processors/FileReaderProcessor.ts | 4 +- .../processors/GithubReaderProcessor.ts | 4 +- .../src/ingestion/processors/YamlProcessor.ts | 4 +- .../src/ingestion/processors/types.ts | 12 +++--- 6 files changed, 35 insertions(+), 33 deletions(-) diff --git a/plugins/catalog-backend/src/ingestion/LocationReaders.ts b/plugins/catalog-backend/src/ingestion/LocationReaders.ts index 708c3e11aa..e806ecfbdb 100644 --- a/plugins/catalog-backend/src/ingestion/LocationReaders.ts +++ b/plugins/catalog-backend/src/ingestion/LocationReaders.ts @@ -16,6 +16,7 @@ import { getVoidLogger } from '@backstage/backend-common'; import { + Entity, EntityPolicies, EntityPolicy, LocationSpec, @@ -29,11 +30,11 @@ import * as result from './processors/results'; import { LocationProcessor, LocationProcessorDataResult, + LocationProcessorEmit, LocationProcessorEntityResult, LocationProcessorErrorResult, LocationProcessorLocationResult, LocationProcessorResult, - LocationProcessorSink, } from './processors/types'; import { YamlProcessor } from './processors/YamlProcessor'; import { LocationReader, ReadLocationResult } from './types'; @@ -74,17 +75,25 @@ export class LocationReaders implements LocationReader { for (let depth = 0; depth < MAX_DEPTH; ++depth) { const newItems: LocationProcessorResult[] = []; - const sink: LocationProcessorSink = i => newItems.push(i); + const emit: LocationProcessorEmit = i => newItems.push(i); for (const item of items) { if (item.type === 'location') { - await this.handleLocation(item, sink); + await this.handleLocation(item, emit); } else if (item.type === 'data') { - await this.handleData(item, sink); + await this.handleData(item, emit); } else if (item.type === 'entity') { - await this.handleEntity(item, sink, output); + const entity = await this.handleEntity(item, emit); + output.entities.push({ + entity, + location: item.location, + }); } else if (item.type === 'error') { - await this.handleError(item, sink, output); + await this.handleError(item, emit); + output.errors.push({ + location: item.location, + error: item.error, + }); } } @@ -103,7 +112,7 @@ export class LocationReaders implements LocationReader { private async handleLocation( item: LocationProcessorLocationResult, - emit: LocationProcessorSink, + emit: LocationProcessorEmit, ) { this.logger.debug( `Reading location ${item.location.type} ${item.location.target} optional=${item.optional}`, @@ -130,7 +139,7 @@ export class LocationReaders implements LocationReader { private async handleData( item: LocationProcessorDataResult, - emit: LocationProcessorSink, + emit: LocationProcessorEmit, ) { this.logger.debug( `Parsing data from location ${item.location.type} ${item.location.target} (${item.data.byteLength} bytes)`, @@ -155,9 +164,8 @@ export class LocationReaders implements LocationReader { private async handleEntity( item: LocationProcessorEntityResult, - emit: LocationProcessorSink, - output: ReadLocationResult, - ) { + emit: LocationProcessorEmit, + ): Promise { this.logger.debug( `Got entity at location ${item.location.type} ${item.location.target}, ${item.entity.apiVersion} ${item.entity.kind}`, ); @@ -175,13 +183,12 @@ export class LocationReaders implements LocationReader { } } - output.entities.push({ entity: current, location: item.location }); + return current; } private async handleError( item: LocationProcessorErrorResult, - emit: LocationProcessorSink, - output: ReadLocationResult, + emit: LocationProcessorEmit, ) { this.logger.debug( `Encountered error at location ${item.location.type} ${item.location.target}, ${item.error}`, @@ -197,10 +204,5 @@ export class LocationReaders implements LocationReader { } } } - - output.errors.push({ - location: item.location, - error: item.error, - }); } } diff --git a/plugins/catalog-backend/src/ingestion/index.ts b/plugins/catalog-backend/src/ingestion/index.ts index dc9bce56f2..07e917b97b 100644 --- a/plugins/catalog-backend/src/ingestion/index.ts +++ b/plugins/catalog-backend/src/ingestion/index.ts @@ -17,10 +17,10 @@ export { HigherOrderOperations } from './HigherOrderOperations'; export { LocationReaders } from './LocationReaders'; export type { - HigherOrderOperation, AddLocationResult, + HigherOrderOperation, LocationReader, - ReadLocationResult, ReadLocationEntity, ReadLocationError, + ReadLocationResult, } from './types'; diff --git a/plugins/catalog-backend/src/ingestion/processors/FileReaderProcessor.ts b/plugins/catalog-backend/src/ingestion/processors/FileReaderProcessor.ts index a3d24575bc..b95e2ddb58 100644 --- a/plugins/catalog-backend/src/ingestion/processors/FileReaderProcessor.ts +++ b/plugins/catalog-backend/src/ingestion/processors/FileReaderProcessor.ts @@ -17,13 +17,13 @@ import { LocationSpec } from '@backstage/catalog-model'; import fs from 'fs-extra'; import * as result from './results'; -import { LocationProcessor, LocationProcessorSink } from './types'; +import { LocationProcessor, LocationProcessorEmit } from './types'; export class FileReaderProcessor implements LocationProcessor { async readLocation( location: LocationSpec, optional: boolean, - emit: LocationProcessorSink, + emit: LocationProcessorEmit, ): Promise { if (location.type !== 'file') { return false; diff --git a/plugins/catalog-backend/src/ingestion/processors/GithubReaderProcessor.ts b/plugins/catalog-backend/src/ingestion/processors/GithubReaderProcessor.ts index a39380027b..b83c9a16f3 100644 --- a/plugins/catalog-backend/src/ingestion/processors/GithubReaderProcessor.ts +++ b/plugins/catalog-backend/src/ingestion/processors/GithubReaderProcessor.ts @@ -17,13 +17,13 @@ import { LocationSpec } from '@backstage/catalog-model'; import fetch from 'node-fetch'; import * as result from './results'; -import { LocationProcessor, LocationProcessorSink } from './types'; +import { LocationProcessor, LocationProcessorEmit } from './types'; export class GithubReaderProcessor implements LocationProcessor { async readLocation( location: LocationSpec, optional: boolean, - emit: LocationProcessorSink, + emit: LocationProcessorEmit, ): Promise { if (location.type !== 'github') { return false; diff --git a/plugins/catalog-backend/src/ingestion/processors/YamlProcessor.ts b/plugins/catalog-backend/src/ingestion/processors/YamlProcessor.ts index 1faf77646a..6a2b5cf419 100644 --- a/plugins/catalog-backend/src/ingestion/processors/YamlProcessor.ts +++ b/plugins/catalog-backend/src/ingestion/processors/YamlProcessor.ts @@ -18,13 +18,13 @@ import { Entity, LocationSpec } from '@backstage/catalog-model'; import lodash from 'lodash'; import yaml from 'yaml'; import * as result from './results'; -import { LocationProcessor, LocationProcessorSink } from './types'; +import { LocationProcessor, LocationProcessorEmit } from './types'; export class YamlProcessor implements LocationProcessor { async parseData( data: Buffer, location: LocationSpec, - emit: LocationProcessorSink, + emit: LocationProcessorEmit, ): Promise { if (!location.target.match(/\.ya?ml$/)) { return false; diff --git a/plugins/catalog-backend/src/ingestion/processors/types.ts b/plugins/catalog-backend/src/ingestion/processors/types.ts index 43e5480029..c8f3f6f482 100644 --- a/plugins/catalog-backend/src/ingestion/processors/types.ts +++ b/plugins/catalog-backend/src/ingestion/processors/types.ts @@ -14,7 +14,7 @@ * limitations under the License. */ -import { LocationSpec, Entity } from '@backstage/catalog-model'; +import { Entity, LocationSpec } from '@backstage/catalog-model'; export type LocationProcessor = { /** @@ -28,7 +28,7 @@ export type LocationProcessor = { readLocation?( location: LocationSpec, optional: boolean, - emit: LocationProcessorSink, + emit: LocationProcessorEmit, ): Promise; /** @@ -42,7 +42,7 @@ export type LocationProcessor = { parseData?( data: Buffer, location: LocationSpec, - emit: LocationProcessorSink, + emit: LocationProcessorEmit, ): Promise; /** @@ -56,7 +56,7 @@ export type LocationProcessor = { processEntity?( entity: Entity, location: LocationSpec, - emit: LocationProcessorSink, + emit: LocationProcessorEmit, ): Promise; /** @@ -70,11 +70,11 @@ export type LocationProcessor = { handleError?( error: Error, location: LocationSpec, - emit: LocationProcessorSink, + emit: LocationProcessorEmit, ): Promise; }; -export type LocationProcessorSink = ( +export type LocationProcessorEmit = ( generated: LocationProcessorResult, ) => void;