catalog-backend: implement caching in the processing orchestrator

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

plugins/catalog-backend/src/ingestion/LocationReaders.ts

@@ -52,6 +52,13 @@ type Options = {
   policy: EntityPolicy;
 };
 
+const noopCache = {
+  async get() {
+    return undefined;
+  },
+  async set() {},
+};
+
 /**
  * Implements the reading of a location through a series of processor tasks.
  */
@@ -167,6 +174,7 @@ export class LocationReaders implements LocationReader {
               item.optional,
               validatedEmit,
               this.options.parser,
+              noopCache,
             )
           ) {
             return;
@@ -215,6 +223,7 @@ export class LocationReaders implements LocationReader {
             item.location,
             emit,
             originLocation,
+            noopCache,
           );
         } catch (e) {
           const message = `Processor ${
@@ -285,6 +294,7 @@ export class LocationReaders implements LocationReader {
             current,
             item.location,
             emit,
+            noopCache,
           );
         } catch (e) {
           const message = `Processor ${

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

@@ -44,6 +44,10 @@ type CacheItem = {
 export class UrlReaderProcessor implements CatalogProcessor {
   constructor(private readonly options: Options) {}
 
+  getProcessorName() {
+    return 'url-reader';
+  }
+
   async readLocation(
     location: LocationSpec,
     optional: boolean,
@@ -74,13 +78,11 @@ export class UrlReaderProcessor implements CatalogProcessor {
         }
       }
 
-      const isOnlyEntities = parseResults.every(
-        (r): r is CatalogProcessorEntityResult => r.type === 'entity',
-      );
+      const isOnlyEntities = parseResults.every(r => r.type === 'entity');
       if (newEtag && isOnlyEntities) {
         await cache.set<CacheItem>(CACHE_KEY, {
           etag: newEtag,
-          value: parseResults,
+          value: parseResults as CatalogProcessorEntityResult[],
         });
       }
     } catch (error) {
@@ -121,7 +123,7 @@ export class UrlReaderProcessor implements CatalogProcessor {
 
     // Otherwise do a plain read, prioritizing readUrl if available
     if (this.options.reader.readUrl) {
-      const data = await this.options.reader.readUrl(location);
+      const data = await this.options.reader.readUrl(location, { etag });
       return [[{ url: location, data: await data.buffer() }], data.etag];
     }
 

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

@@ -22,6 +22,12 @@ import {
 import { JsonValue } from '@backstage/config';
 
 export type CatalogProcessor = {
+  /**
+   * A unique identifier for the Catalog Processor.
+   * It's strongly recommended implement getProcessorName as this method will be required in the future.
+   */
+  getProcessorName?(): string;
+
   /**
    * Reads the contents of a location.
    *

plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.test.ts

@@ -55,7 +55,7 @@ describe('DefaultCatalogProcessingEngine', () => {
       relations: [],
       errors: [],
       deferredEntities: [],
-      state: new Map(),
+      state: {},
     });
     const engine = new DefaultCatalogProcessingEngine(
       getVoidLogger(),
@@ -84,7 +84,7 @@ describe('DefaultCatalogProcessingEngine', () => {
               metadata: { name: 'test' },
             },
             resultHash: '',
-            state: new Map(),
+            state: {},
             nextUpdateAt: DateTime.now(),
             lastDiscoveryAt: DateTime.now(),
           },
@@ -117,7 +117,7 @@ describe('DefaultCatalogProcessingEngine', () => {
       relations: [],
       errors: [],
       deferredEntities: [],
-      state: new Map(),
+      state: {},
     });
     const engine = new DefaultCatalogProcessingEngine(
       getVoidLogger(),
@@ -147,7 +147,7 @@ describe('DefaultCatalogProcessingEngine', () => {
               metadata: { name: 'test' },
             },
             resultHash: '',
-            state: new Map(),
+            state: {},
             nextUpdateAt: DateTime.now(),
             lastDiscoveryAt: DateTime.now(),
           },
@@ -181,7 +181,7 @@ describe('DefaultCatalogProcessingEngine', () => {
       entityRef: '',
       unprocessedEntity: entity,
       resultHash: 'the matching hash',
-      state: new Map(),
+      state: {},
       nextUpdateAt: DateTime.now(),
       lastDiscoveryAt: DateTime.now(),
     };
@@ -194,7 +194,7 @@ describe('DefaultCatalogProcessingEngine', () => {
       relations: [],
       errors: [],
       deferredEntities: [],
-      state: new Map(),
+      state: {},
     });
 
     const engine = new DefaultCatalogProcessingEngine(

plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts

@@ -167,8 +167,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
             hashBuilder = hashBuilder
               .update(stableStringify({ ...result.completedEntity }))
               .update(stableStringify([...result.deferredEntities]))
-              .update(stableStringify([...result.relations]))
-              .update(stableStringify(Object.fromEntries(result.state)));
+              .update(stableStringify([...result.relations]));
           }
 
           const resultHash = hashBuilder.digest('hex');

plugins/catalog-backend/src/next/DefaultLocationService.test.ts

@@ -39,7 +39,7 @@ describe('DefaultLocationServiceTest', () => {
       store.listLocations.mockResolvedValueOnce([]);
       orchestrator.process.mockResolvedValueOnce({
         ok: true,
-        state: new Map(),
+        state: {},
         completedEntity: {
           apiVersion: 'backstage.io/v1alpha1',
           kind: 'Location',
@@ -65,7 +65,7 @@ describe('DefaultLocationServiceTest', () => {
 
       orchestrator.process.mockResolvedValueOnce({
         ok: true,
-        state: new Map(),
+        state: {},
         completedEntity: {
           apiVersion: 'backstage.io/v1alpha1',
           kind: 'Component',

plugins/catalog-backend/src/next/DefaultLocationService.ts

@@ -87,7 +87,6 @@ export class DefaultLocationService implements LocationService {
       { entity, locationKey: `${spec.type}:${spec.target}` },
     ];
     const entities: Entity[] = [];
-    const state = new Map(); // ignored
     while (unprocessedEntities.length) {
       const currentEntity = unprocessedEntities.pop();
       if (!currentEntity) {
@@ -95,7 +94,7 @@ export class DefaultLocationService implements LocationService {
       }
       const processed = await this.orchestrator.process({
         entity: currentEntity.entity,
-        state,
+        state: {}, // we process without the existing cache
       });
 
       if (processed.ok) {

plugins/catalog-backend/src/next/DefaultRefreshService.test.ts

@@ -138,7 +138,7 @@ describe('Refresh integration', () => {
             relations: [],
             errors: [],
             deferredEntities,
-            state: new Map(),
+            state: {},
           };
         },
       },

plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts

@@ -17,7 +17,6 @@
 import { getVoidLogger } from '@backstage/backend-common';
 import { TestDatabaseId, TestDatabases } from '@backstage/backend-test-utils';
 import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
-import { JsonObject } from '@backstage/config';
 import { Knex } from 'knex';
 import * as uuid from 'uuid';
 import { Logger } from 'winston';
@@ -213,7 +212,7 @@ describe('Default Processing Database', () => {
               id,
               processedEntity,
               resultHash: '',
-              state: new Map<string, JsonObject>(),
+              state: {},
               relations: [],
               deferredEntities: [],
             }),
@@ -232,7 +231,7 @@ describe('Default Processing Database', () => {
           id,
           processedEntity,
           resultHash: '',
-          state: new Map<string, JsonObject>(),
+          state: {},
           relations: [],
           deferredEntities: [],
           locationKey: 'key',
@@ -285,8 +284,7 @@ describe('Default Processing Database', () => {
           last_discovery_at: '2021-04-01 13:37:00',
         });
 
-        const state = new Map<string, JsonObject>();
-        state.set('hello', { t: 'something' });
+        const state = { hello: { t: 'something' } };
 
         await db.transaction(tx =>
           db.updateProcessedEntity(tx, {
@@ -308,9 +306,7 @@ describe('Default Processing Database', () => {
         expect(entities[0].processed_entity).toEqual(
           JSON.stringify(processedEntity),
         );
-        expect(entities[0].cache).toEqual(
-          JSON.stringify(Object.fromEntries(state)),
-        );
+        expect(entities[0].cache).toEqual(JSON.stringify(state));
         expect(entities[0].errors).toEqual("['something broke']");
         expect(entities[0].location_key).toEqual('key');
       },
@@ -352,7 +348,7 @@ describe('Default Processing Database', () => {
             id,
             processedEntity,
             resultHash: '',
-            state: new Map<string, JsonObject>(),
+            state: {},
             relations: relations,
             deferredEntities: [],
           }),
@@ -404,7 +400,7 @@ describe('Default Processing Database', () => {
             id,
             processedEntity,
             resultHash: '',
-            state: new Map<string, JsonObject>(),
+            state: {},
             relations: [],
             deferredEntities,
           }),

plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts

@@ -15,7 +15,6 @@
  */
 
 import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
-import { JsonObject } from '@backstage/config';
 import { ConflictError, NotFoundError } from '@backstage/errors';
 import { Knex } from 'knex';
 import lodash from 'lodash';
@@ -80,7 +79,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
       .update({
         processed_entity: JSON.stringify(processedEntity),
         result_hash: resultHash,
-        cache: JSON.stringify(Object.fromEntries(state || [])),
+        cache: JSON.stringify(state),
         errors,
         location_key: locationKey,
       })
@@ -508,9 +507,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
             resultHash: i.result_hash || '',
             nextUpdateAt: timestampToDateTime(i.next_update_at),
             lastDiscoveryAt: timestampToDateTime(i.last_discovery_at),
-            state: i.cache
-              ? JSON.parse(i.cache)
-              : new Map<string, JsonObject>(),
+            state: i.cache ? JSON.parse(i.cache) : undefined,
             errors: i.errors,
             locationKey: i.location_key,
           } as RefreshStateItem),

plugins/catalog-backend/src/next/database/types.ts

@@ -15,7 +15,7 @@
  */
 
 import { Entity, EntityRelationSpec } from '@backstage/catalog-model';
-import { JsonObject } from '@backstage/config';
+import { JsonValue } from '@backstage/config';
 import { DateTime } from 'luxon';
 import { Transaction } from '../../database/types';
 import { DeferredEntity } from '../processing/types';
@@ -36,7 +36,7 @@ export type UpdateProcessedEntityOptions = {
   id: string;
   processedEntity: Entity;
   resultHash: string;
-  state?: Map<string, JsonObject>;
+  state?: JsonValue;
   errors?: string;
   relations: EntityRelationSpec[];
   deferredEntities: DeferredEntity[];
@@ -57,7 +57,7 @@ export type RefreshStateItem = {
   resultHash: string;
   nextUpdateAt: DateTime;
   lastDiscoveryAt: DateTime; // remove?
-  state: Map<string, JsonObject>;
+  state?: JsonValue;
   errors?: string;
   locationKey?: string;
 };

plugins/catalog-backend/src/next/processing/DefaultCatalogProcessingOrchestrator.ts

@@ -24,6 +24,7 @@ import {
   stringifyLocationReference,
 } from '@backstage/catalog-model';
 import { ConflictError, InputError, NotAllowedError } from '@backstage/errors';
+import { JsonObject, JsonValue } from '@backstage/config';
 import { ScmIntegrationRegistry } from '@backstage/integration';
 import path from 'path';
 import { Logger } from 'winston';
@@ -32,6 +33,7 @@ import {
   CatalogProcessorParser,
 } from '../../ingestion/processors';
 import * as results from '../../ingestion/processors/results';
+import { CatalogProcessorCache } from '../../ingestion/processors/types';
 import {
   CatalogProcessingOrchestrator,
   EntityProcessingRequest,
@@ -53,8 +55,70 @@ type Context = {
   location: LocationSpec;
   originLocation: LocationSpec;
   collector: ProcessorOutputCollector;
+  cache: ProcessorCache;
 };
 
+function isObject(value: JsonValue | undefined): value is JsonObject {
+  return typeof value === 'object' && value !== null && !Array.isArray(value);
+}
+
+class DefaultCatalogProcessorCache implements CatalogProcessorCache {
+  private newState?: JsonObject;
+  constructor(private readonly existingState: JsonObject) {}
+
+  async get<ItemType extends JsonValue>(
+    key: string,
+  ): Promise<ItemType | undefined> {
+    return this.existingState[key] as ItemType | undefined;
+  }
+
+  async set<ItemType extends JsonValue>(
+    key: string,
+    value: ItemType,
+  ): Promise<void> {
+    if (!this.newState) {
+      this.newState = {};
+    }
+
+    this.newState[key] = value;
+  }
+
+  collect(): JsonObject | undefined {
+    return this.newState;
+  }
+}
+
+class ProcessorCache {
+  private caches = new Map<string, DefaultCatalogProcessorCache>();
+
+  constructor(private readonly existingState: JsonObject) {}
+  forProcessor(processor: CatalogProcessor): CatalogProcessorCache {
+    // constructor name will be deprecated in the future when we make `getProcessorName` required in the implementation
+    const name = processor.getProcessorName?.() ?? processor.constructor.name;
+    const cache = this.caches.get(name);
+    if (cache) {
+      return cache;
+    }
+
+    const existing = this.existingState[name];
+
+    const newCache = new DefaultCatalogProcessorCache(
+      isObject(existing) ? existing : {},
+    );
+    this.caches.set(name, newCache);
+    return newCache;
+  }
+
+  collect(): JsonObject {
+    const result: JsonObject = {};
+    for (const [key, value] of this.caches.entries()) {
+      result[key] = value.collect();
+    }
+
+    return result;
+  }
+}
+
 export class DefaultCatalogProcessingOrchestrator
   implements CatalogProcessingOrchestrator
 {
@@ -72,17 +136,23 @@ export class DefaultCatalogProcessingOrchestrator
   async process(
     request: EntityProcessingRequest,
   ): Promise<EntityProcessingResult> {
-    return this.processSingleEntity(request.entity);
+    return this.processSingleEntity(request.entity, request.state);
   }
 
   private async processSingleEntity(
     unprocessedEntity: Entity,
+    state: JsonValue | undefined,
   ): Promise<EntityProcessingResult> {
     const collector = new ProcessorOutputCollector(
       this.options.logger,
       unprocessedEntity,
     );
 
+    // Cache that is scoped to the entity and processor
+    const cache = new ProcessorCache(
+      isObject(state) && isObject(state.cache) ? state.cache : {},
+    );
+
     try {
       // This will be checked and mutated step by step below
       let entity: Entity = unprocessedEntity;
@@ -108,6 +178,7 @@ export class DefaultCatalogProcessingOrchestrator
         originLocation: parseLocationReference(
           getEntityOriginLocationRef(entity),
         ),
+        cache,
         collector,
       };
 
@@ -145,7 +216,7 @@ export class DefaultCatalogProcessingOrchestrator
       return {
         ...collectorResults,
         completedEntity: entity,
-        state: new Map(),
+        state: { cache: cache.collect() },
         ok: true,
       };
     } catch (error) {
@@ -173,6 +244,7 @@ export class DefaultCatalogProcessingOrchestrator
             context.location,
             context.collector.onEmit,
             context.originLocation,
+            context.cache.forProcessor(processor),
           );
         } catch (e) {
           throw new InputError(
@@ -306,6 +378,7 @@ export class DefaultCatalogProcessingOrchestrator
               false,
               context.collector.onEmit,
               this.options.parser,
+              context.cache.forProcessor(processor),
             );
             if (read) {
               didRead = true;
@@ -343,6 +416,7 @@ export class DefaultCatalogProcessingOrchestrator
             result,
             context.location,
             context.collector.onEmit,
+            context.cache.forProcessor(processor),
           );
         } catch (e) {
           throw new InputError(

plugins/catalog-backend/src/next/processing/types.ts

@@ -15,17 +15,17 @@
  */
 
 import { Entity, EntityRelationSpec } from '@backstage/catalog-model';
-import { JsonObject } from '@backstage/config';
+import { JsonValue } from '@backstage/config';
 
 export type EntityProcessingRequest = {
   entity: Entity;
-  state: Map<string, JsonObject>; // Versions for multiple deployments etc
+  state?: JsonValue; // Versions for multiple deployments etc
 };
 
 export type EntityProcessingResult =
   | {
       ok: true;
-      state: Map<string, JsonObject>;
+      state: JsonValue;
       completedEntity: Entity;
       deferredEntities: DeferredEntity[];
       relations: EntityRelationSpec[];