Switch to using generators

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,
   );
 

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);

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<ReadLocationResult> {
-    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<void> {
-    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<void> {
-    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<void> {
-    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<void> {
-    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<void>,
+    location: LocationSpec,
+    stopAfterFirstHandled: boolean,
+    failIfNotHandled: boolean,
+  ): Promise<void> {
+    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) });
+      }
+    }
+  }
 }

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<Entity> {
-    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);
   }
 }

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<Entity> {
-    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());
+    }
   }
 }

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<LocationProcessorResult[] | undefined> {
+    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);
     }
   }
 }

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<LocationProcessorResult[] | undefined> {
+  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);
     }
   }
 

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<LocationProcessorResult[] | undefined> {
+  ): 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 };
-    });
+    }
   }
 }

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 };
+}

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<LocationProcessorResult[] | undefined>;
+    optional: boolean,
+  ): LocationProcessorResults;
 
-  parseData?(
-    data: Buffer,
+  parseData?(data: Buffer, location: LocationSpec): LocationProcessorResults;
+
+  processEntity?(
+    entity: Entity,
     location: LocationSpec,
-  ): Promise<LocationProcessorResult[] | undefined>;
+  ): LocationProcessorResults;
 
-  processEntity?(entity: Entity, location: LocationSpec): Promise<Entity>;
-
-  handleError?(error: Error, location: LocationSpec): Promise<void>;
+  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

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,
   );