Get rid of the generators

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

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

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<ReadLocationResult> {
     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<void> {
-    // 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<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;
+      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,
+    });
   }
 }

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<Entity> {
+    return lodash.merge(
       {
         metadata: {
           annotations: {
@@ -34,6 +30,5 @@ export class AnnotateLocationEntityProcessor implements LocationProcessor {
       },
       entity,
     );
-    yield result.entity(location, merged);
   }
 }

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<Entity> {
+    return await this.policy.enforce(entity);
   }
 }

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<boolean> {
     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;
   }
 }

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<boolean> {
     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

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<boolean> {
     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;
   }
 }

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<boolean>;
 
-  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<boolean>;
 
+  /**
+   * 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<Entity>;
 
-  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<void>;
 };
 
-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;