catalog-backend: inital implementation of caching in UrlReaderProcessor + types

Co-authored-by: Fredrik Adelöw <freben@gmail.com>
Co-authored-by: blam <ben@blam.sh>
Co-authored-by: Johan Haals <johan.haals@gmail.com>
Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>

plugins/catalog-backend/src/ingestion/processors/UrlReaderProcessor.ts

@@ -22,15 +22,25 @@ import { Logger } from 'winston';
 import * as result from './results';
 import {
   CatalogProcessor,
+  CatalogProcessorCache,
   CatalogProcessorEmit,
+  CatalogProcessorEntityResult,
   CatalogProcessorParser,
+  CatalogProcessorResult,
 } from './types';
 
+const CACHE_KEY = 'v1';
+
 type Options = {
   reader: UrlReader;
   logger: Logger;
 };
 
+type CacheItem = {
+  etag: string;
+  value: CatalogProcessorEntityResult[];
+};
+
 export class UrlReaderProcessor implements CatalogProcessor {
   constructor(private readonly options: Options) {}
 
@@ -39,25 +49,48 @@ export class UrlReaderProcessor implements CatalogProcessor {
     optional: boolean,
     emit: CatalogProcessorEmit,
     parser: CatalogProcessorParser,
+    cache: CatalogProcessorCache,
   ): Promise<boolean> {
     if (location.type !== 'url') {
       return false;
     }
 
+    const cacheItem = await cache.get<CacheItem>(CACHE_KEY);
+
     try {
-      const output = await this.doRead(location.target);
+      const [output, newEtag] = await this.doRead(
+        location.target,
+        cacheItem?.etag,
+      );
+
+      const parseResults: CatalogProcessorResult[] = [];
       for (const item of output) {
         for await (const parseResult of parser({
           data: item.data,
           location: { type: location.type, target: item.url },
         })) {
+          parseResults.push(parseResult);
           emit(parseResult);
         }
       }
+
+      const isOnlyEntities = parseResults.every(
+        (r): r is CatalogProcessorEntityResult => r.type === 'entity',
+      );
+      if (newEtag && isOnlyEntities) {
+        await cache.set<CacheItem>(CACHE_KEY, {
+          etag: newEtag,
+          value: parseResults,
+        });
+      }
     } catch (error) {
       const message = `Unable to read ${location.type}, ${error}`;
-
-      if (error.name === 'NotFoundError') {
+      if (error.name === 'NotModifiedError' && cacheItem) {
+        for (const parseResult of cacheItem.value) {
+          emit(parseResult);
+        }
+        cache.set<CacheItem>(CACHE_KEY, cacheItem);
+      } else if (error.name === 'NotFoundError') {
         if (!optional) {
           emit(result.notFoundError(location, message));
         }
@@ -71,27 +104,28 @@ export class UrlReaderProcessor implements CatalogProcessor {
 
   private async doRead(
     location: string,
-  ): Promise<{ data: Buffer; url: string }[]> {
+    etag?: string,
+  ): Promise<[response: { data: Buffer; url: string }[], etag?: string]> {
     // Does it contain globs? I.e. does it contain asterisks or question marks
     // (no curly braces for now)
     const { filepath } = parseGitUrl(location);
     if (filepath?.match(/[*?]/)) {
       const limiter = limiterFactory(5);
-      const response = await this.options.reader.search(location);
+      const response = await this.options.reader.search(location, { etag });
       const output = response.files.map(async file => ({
         url: file.url,
         data: await limiter(file.content),
       }));
-      return Promise.all(output);
+      return [await Promise.all(output), response.etag];
     }
 
     // Otherwise do a plain read, prioritizing readUrl if available
     if (this.options.reader.readUrl) {
       const data = await this.options.reader.readUrl(location);
-      return [{ url: location, data: await data.buffer() }];
+      return [[{ url: location, data: await data.buffer() }], data.etag];
     }
 
     const data = await this.options.reader.read(location);
-    return [{ url: location, data }];
+    return [[{ url: location, data }]];
   }
 }

plugins/catalog-backend/src/ingestion/processors/types.ts

@@ -19,16 +19,18 @@ import {
   EntityRelationSpec,
   LocationSpec,
 } from '@backstage/catalog-model';
+import { JsonValue } from '@backstage/config';
 
 export type CatalogProcessor = {
   /**
    * 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
-   * @param parser A parser, that is able to take the raw catalog descriptor
+   * @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
+   * @param parser - A parser, that is able to take the raw catalog descriptor
    *               data and turn it into the actual result pieces.
+   * @param cache - A cache for storing values local to this processor and the current entity.
    * @returns True if handled by this processor, false otherwise
    */
   readLocation?(
@@ -36,6 +38,7 @@ export type CatalogProcessor = {
     optional: boolean,
     emit: CatalogProcessorEmit,
     parser: CatalogProcessorParser,
+    cache: CatalogProcessorCache,
   ): Promise<boolean>;
 
   /**
@@ -46,12 +49,13 @@ export type CatalogProcessor = {
    * additional data, and the input entity may actually still be incomplete
    * when the processor is invoked.
    *
-   * @param entity The (possibly partial) entity to process
-   * @param location The location that the entity came from
-   * @param emit A sink for auxiliary items resulting from the processing
-   * @param originLocation The location that the entity originally came from.
+   * @param entity - The (possibly partial) entity to process
+   * @param location - The location that the entity came from
+   * @param emit - A sink for auxiliary items resulting from the processing
+   * @param originLocation - The location that the entity originally came from.
    *   While location resolves to the direct parent location, originLocation
    *   tells which location was used to start the ingestion loop.
+   * @param cache - A cache for storing values local to this processor and the current entity.
    * @returns The same entity or a modified version of it
    */
   preProcessEntity?(
@@ -59,13 +63,14 @@ export type CatalogProcessor = {
     location: LocationSpec,
     emit: CatalogProcessorEmit,
     originLocation: LocationSpec,
+    cache: CatalogProcessorCache,
   ): Promise<Entity>;
 
   /**
    * Validates the entity as a known entity kind, after it has been pre-
    * processed and has passed through basic overall validation.
    *
-   * @param entity The entity to validate
+   * @param entity - The entity to validate
    * @returns Resolves to true, if the entity was of a kind that was known and
    *   handled by this processor, and was found to be valid. Resolves to false,
    *   if the entity was not of a kind that was known by this processor.
@@ -77,23 +82,25 @@ export type CatalogProcessor = {
   /**
    * Post-processes an emitted entity, after it has been validated.
    *
-   * @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
+   * @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
+   * @param cache - A cache for storing values local to this processor and the current entity.
    * @returns The same entity or a modified version of it
    */
   postProcessEntity?(
     entity: Entity,
     location: LocationSpec,
     emit: CatalogProcessorEmit,
+    cache: CatalogProcessorCache,
   ): Promise<Entity>;
 
   /**
    * 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 handling
+   * @param error - The error
+   * @param location - The location where the error occurred
+   * @param emit - A sink for items resulting from this handling
    * @returns Nothing
    */
   handleError?(
@@ -113,6 +120,34 @@ export type CatalogProcessorParser = (options: {
   location: LocationSpec;
 }) => AsyncIterable<CatalogProcessorResult>;
 
+/**
+ * A cache for storing data during processing.
+ *
+ * The values stored in the cache are always local to each processor, meaning
+ * no processor can see cache values from other processors.
+ *
+ * The cache instance provided to the CatalogProcessor is also scoped to the
+ * entity being processed, meaning that each processor run can't see cache
+ * values from processing runs for other entities.
+ *
+ * Values that are set during a processing run will only be visible in the directly
+ * following run. The cache will be overwritten every run, meaning existing values
+ * are removed and need to be set again for them to remain in the cache.
+ *
+ * @public
+ */
+export interface CatalogProcessorCache {
+  /**
+   * Retrieve a value from the cache.
+   */
+  get<ItemType extends JsonValue>(key: string): Promise<ItemType | undefined>;
+
+  /**
+   * Store a value in the cache.
+   */
+  set<ItemType extends JsonValue>(key: string, value: ItemType): Promise<void>;
+}
+
 export type CatalogProcessorEmit = (generated: CatalogProcessorResult) => void;
 
 export type CatalogProcessorLocationResult = {