Catalog: Add location key to handle duplicate entities

Signed-off-by: Johan Haals <johan.haals@gmail.com>

plugins/catalog-backend/migrations/20210302150147_refresh_state.js

@@ -65,6 +65,8 @@ exports.up = async function up(knex) {
       .dateTime('last_discovery_at') // TODO: timezone or change to epoch-millis or similar
       .notNullable()
       .comment('The last timestamp of which this entity was discovered');
+    // TODO: get migrations to work for this field instead. should be removed before merge.
+    table.text('location_key').nullable().comment('Location key');
     table.unique(['entity_ref'], 'refresh_state_entity_ref_uniq');
     table.index('entity_id', 'refresh_state_entity_id_idx');
     table.index('entity_ref', 'refresh_state_entity_ref_idx');

plugins/catalog-backend/migrations/20210622104022_refresh_state_location_key.js

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2020 The Backstage Authors
+ *
+ * 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.
+ */
+
+// @ts-check
+
+/**
+ * @param {import('knex').Knex} knex
+ */
+exports.up = async function up(knex) {
+  await knex.schema.alterTable('refresh_state', table => {
+    table.text('location_key').nullable().comment('').alter();
+  });
+};
+
+/**
+ * @param {import('knex').Knex} knex
+ */
+exports.down = async function down(knex) {
+  await knex.schema.alterTable('refresh_state', table => {
+    table.dropColumn('location_key');
+  });
+};

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

@@ -16,6 +16,7 @@
 
 import { Config } from '@backstage/config';
 import path from 'path';
+import { getEntityLocationRef } from './processing/util';
 import { EntityProvider, EntityProviderConnection } from './types';
 import { locationSpecToLocationEntity } from './util';
 
@@ -37,10 +38,12 @@ export class ConfigLocationEntityProvider implements EntityProvider {
     const entities = locationConfigs.map(location => {
       const type = location.getString('type');
       const target = location.getString('target');
-      return locationSpecToLocationEntity({
+      const entity = locationSpecToLocationEntity({
         type,
         target: type === 'file' ? path.resolve(target) : target,
       });
+      const emitKey = getEntityLocationRef(entity);
+      return { entity, emitKey };
     });
 
     await this.connection.applyMutation({

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

@@ -46,7 +46,7 @@ class Connection implements EntityProviderConnection {
     const db = this.config.processingDatabase;
 
     if (mutation.type === 'full') {
-      this.check(mutation.entities);
+      this.check(mutation.entities.map(e => e.entity));
       await db.transaction(async tx => {
         await db.replaceUnprocessedEntities(tx, {
           sourceKey: this.config.id,
@@ -57,8 +57,8 @@ class Connection implements EntityProviderConnection {
       return;
     }
 
-    this.check(mutation.added);
-    this.check(mutation.removed);
+    this.check(mutation.added.map(e => e.entity));
+    this.check(mutation.removed.map(e => e.entity));
     await db.transaction(async tx => {
       await db.replaceUnprocessedEntities(tx, {
         sourceKey: this.config.id,
@@ -125,7 +125,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
       },
       processTask: async item => {
         try {
-          const { id, state, unprocessedEntity, entityRef } = item;
+          const { id, state, unprocessedEntity, entityRef, emitKey } = item;
           const result = await this.orchestrator.process({
             entity: unprocessedEntity,
             state,
@@ -171,6 +171,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine {
               errors: errorsString,
               relations: result.relations,
               deferredEntities: result.deferredEntities,
+              emitKey,
             });
           });
 

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

@@ -19,6 +19,7 @@ import { ConflictError, NotFoundError } from '@backstage/errors';
 import { Knex } from 'knex';
 import { v4 as uuid } from 'uuid';
 import { DbLocationsRow } from './database/tables';
+import { getEntityLocationRef } from './processing/util';
 import {
   EntityProvider,
   EntityProviderConnection,
@@ -60,10 +61,10 @@ export class DefaultLocationStore implements LocationStore, EntityProvider {
 
       return inner;
     });
-
+    const entity = locationSpecToLocationEntity(location);
     await this.connection.applyMutation({
       type: 'delta',
-      added: [locationSpecToLocationEntity(location)],
+      added: [{ entity, emitKey: getEntityLocationRef(entity) }],
       removed: [],
     });
 
@@ -102,11 +103,11 @@ export class DefaultLocationStore implements LocationStore, EntityProvider {
       await tx<DbLocationsRow>('locations').where({ id }).del();
       return location;
     });
-
+    const entity = locationSpecToLocationEntity(deleted);
     await this.connection.applyMutation({
       type: 'delta',
       added: [],
-      removed: [locationSpecToLocationEntity(deleted)],
+      removed: [{ entity, emitKey: getEntityLocationRef(entity) }],
     });
   }
 
@@ -124,7 +125,8 @@ export class DefaultLocationStore implements LocationStore, EntityProvider {
     const locations = await this.locations();
 
     const entities = locations.map(location => {
-      return locationSpecToLocationEntity(location);
+      const entity = locationSpecToLocationEntity(location);
+      return { entity, emitKey: getEntityLocationRef(entity) };
     });
 
     await this.connection.applyMutation({

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

@@ -22,6 +22,7 @@ import lodash from 'lodash';
 import { v4 as uuid } from 'uuid';
 import type { Logger } from 'winston';
 import { Transaction } from '../../database';
+import { DeferredEntity } from '../processing/types';
 import {
   DbRefreshStateReferencesRow,
   DbRefreshStateRow,
@@ -63,15 +64,18 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
       errors,
       relations,
       deferredEntities,
+      emitKey,
     } = options;
-
     const refreshResult = await tx<DbRefreshStateRow>('refresh_state')
       .update({
         processed_entity: JSON.stringify(processedEntity),
         cache: JSON.stringify(state),
         errors,
+        emit_key: emitKey,
       })
-      .where('entity_id', id);
+      .where('entity_id', id)
+      .andWhere('emit_key', emitKey)
+      .orWhere('emit_key', '');
     if (refreshResult === 0) {
       throw new NotFoundError(`Processing state not found for ${id}`);
     }
@@ -129,11 +133,11 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
   private async createDelta(
     tx: Knex.Transaction,
     options: ReplaceUnprocessedEntitiesOptions,
-  ): Promise<{ toAdd: Entity[]; toRemove: string[] }> {
+  ): Promise<{ toAdd: DeferredEntity[]; toRemove: string[] }> {
     if (options.type === 'delta') {
       return {
         toAdd: options.added,
-        toRemove: options.removed.map(e => stringifyEntityRef(e)),
+        toRemove: options.removed.map(e => stringifyEntityRef(e.entity)),
       };
     }
 
@@ -146,7 +150,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
 
     const items = options.items.map(entity => ({
       entity,
-      ref: stringifyEntityRef(entity),
+      ref: stringifyEntityRef(entity.entity),
     }));
 
     const oldRefsSet = new Set(oldRefs);
@@ -281,19 +285,22 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
     }
 
     if (toAdd.length) {
-      const state: Knex.DbRecord<DbRefreshStateRow>[] = toAdd.map(entity => ({
-        entity_id: uuid(),
-        entity_ref: stringifyEntityRef(entity),
-        unprocessed_entity: JSON.stringify(entity),
-        errors: '',
-        next_update_at: tx.fn.now(),
-        last_discovery_at: tx.fn.now(),
-      }));
+      const state: Knex.DbRecord<DbRefreshStateRow>[] = toAdd.map(
+        ({ entity, emitKey }) => ({
+          entity_id: uuid(),
+          entity_ref: stringifyEntityRef(entity),
+          unprocessed_entity: JSON.stringify(entity),
+          emit_key: emitKey,
+          errors: '',
+          next_update_at: tx.fn.now(),
+          last_discovery_at: tx.fn.now(),
+        }),
+      );
 
       const stateReferences: DbRefreshStateReferencesRow[] = toAdd.map(
         entity => ({
           source_key: options.sourceKey,
-          target_entity_ref: stringifyEntityRef(entity),
+          target_entity_ref: stringifyEntityRef(entity.entity),
         }),
       );
       // TODO(freben): Concurrency? If we did these one by one, a .onConflict().merge would have made sense
@@ -313,12 +320,13 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
     const tx = txOpaque as Knex.Transaction;
 
     const stateRows = options.entities.map(
-      entity =>
+      ({ entity, emitKey }) =>
         ({
           entity_id: uuid(),
           entity_ref: stringifyEntityRef(entity),
           unprocessed_entity: JSON.stringify(entity),
           errors: '',
+          emit_key: emitKey,
           next_update_at: tx.fn.now(),
           last_discovery_at: tx.fn.now(),
         } as Knex.DbRecord<DbRefreshStateRow>),
@@ -406,6 +414,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
               ? JSON.parse(i.cache)
               : new Map<string, JsonObject>(),
             errors: i.errors,
+            emitKey: i.emit_key,
           } as RefreshStateItem),
       ),
     };

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

@@ -29,6 +29,7 @@ export type DbRefreshStateRow = {
   next_update_at: string;
   last_discovery_at: string; // remove?
   errors?: string;
+  emit_key: string;
 };
 
 export type DbRefreshStateReferencesRow = {

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

@@ -17,10 +17,11 @@
 import { Entity, EntityRelationSpec } from '@backstage/catalog-model';
 import { JsonObject } from '@backstage/config';
 import { Transaction } from '../../database/types';
+import { DeferredEntity } from '../processing/types';
 
 export type AddUnprocessedEntitiesOptions = {
   entityRef: string;
-  entities: Entity[];
+  entities: DeferredEntity[];
 };
 
 export type AddUnprocessedEntitiesResult = {};
@@ -31,7 +32,8 @@ export type UpdateProcessedEntityOptions = {
   state?: Map<string, JsonObject>;
   errors?: string;
   relations: EntityRelationSpec[];
-  deferredEntities: Entity[];
+  deferredEntities: DeferredEntity[];
+  emitKey: string;
 };
 
 export type UpdateProcessedEntityErrorsOptions = {
@@ -48,6 +50,7 @@ export type RefreshStateItem = {
   lastDiscoveryAt: string; // remove?
   state: Map<string, JsonObject>;
   errors?: string;
+  emitKey: string;
 };
 
 export type GetProcessableEntitiesResult = {
@@ -57,13 +60,13 @@ export type GetProcessableEntitiesResult = {
 export type ReplaceUnprocessedEntitiesOptions =
   | {
       sourceKey: string;
-      items: Entity[];
+      items: DeferredEntity[];
       type: 'full';
     }
   | {
       sourceKey: string;
-      added: Entity[];
-      removed: Entity[];
+      added: DeferredEntity[];
+      removed: DeferredEntity[];
       type: 'delta';
     };
 

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

@@ -68,7 +68,6 @@ export class DefaultCatalogProcessingOrchestrator
   async process(
     request: EntityProcessingRequest,
   ): Promise<EntityProcessingResult> {
-    // TODO: implement dryRun/eager
     return this.processSingleEntity(request.entity);
   }
 

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

@@ -24,7 +24,12 @@ import {
 import { Logger } from 'winston';
 import { CatalogProcessorResult } from '../../ingestion';
 import { locationSpecToLocationEntity } from '../util';
-import { getEntityOriginLocationRef, validateEntityEnvelope } from './util';
+import { DeferredEntity } from './types';
+import {
+  getEntityLocationRef,
+  getEntityOriginLocationRef,
+  validateEntityEnvelope,
+} from './util';
 
 /**
  * Helper class for aggregating all of the emitted data from processors.
@@ -32,7 +37,7 @@ import { getEntityOriginLocationRef, validateEntityEnvelope } from './util';
 export class ProcessorOutputCollector {
   private readonly errors = new Array<Error>();
   private readonly relations = new Array<EntityRelationSpec>();
-  private readonly deferredEntities = new Array<Entity>();
+  private readonly deferredEntities = new Array<DeferredEntity>();
   private done = false;
 
   constructor(
@@ -73,6 +78,8 @@ export class ProcessorOutputCollector {
         return;
       }
 
+      const location = stringifyLocationReference(i.location);
+
       // Note that at this point, we have only validated the envelope part of
       // the entity data. Annotations are not part of that, so we have to be
       // defensive. If the annotations were malformed (e.g. were not a valid
@@ -81,7 +88,6 @@ export class ProcessorOutputCollector {
       const annotations = entity.metadata.annotations || {};
       if (typeof annotations === 'object' && !Array.isArray(annotations)) {
         const originLocation = getEntityOriginLocationRef(this.parentEntity);
-        const location = stringifyLocationReference(i.location);
         entity = {
           ...entity,
           metadata: {
@@ -95,11 +101,14 @@ export class ProcessorOutputCollector {
         };
       }
 
-      this.deferredEntities.push(entity);
+      this.deferredEntities.push({ entity, emitKey: location });
     } else if (i.type === 'location') {
-      this.deferredEntities.push(
-        locationSpecToLocationEntity(i.location, this.parentEntity),
+      const entity = locationSpecToLocationEntity(
+        i.location,
+        this.parentEntity,
       );
+      const emitKey = getEntityLocationRef(entity);
+      this.deferredEntities.push({ entity, emitKey });
     } else if (i.type === 'relation') {
       this.relations.push(i.relation);
     } else if (i.type === 'error') {

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

@@ -27,7 +27,7 @@ export type EntityProcessingResult =
       ok: true;
       state: Map<string, JsonObject>;
       completedEntity: Entity;
-      deferredEntities: Entity[];
+      deferredEntities: DeferredEntity[];
       relations: EntityRelationSpec[];
       errors: Error[];
     }
@@ -39,3 +39,8 @@ export type EntityProcessingResult =
 export interface CatalogProcessingOrchestrator {
   process(request: EntityProcessingRequest): Promise<EntityProcessingResult>;
 }
+
+export type DeferredEntity = {
+  entity: Entity;
+  emitKey: string;
+};

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

@@ -15,6 +15,7 @@
  */
 
 import { Entity, Location, LocationSpec } from '@backstage/catalog-model';
+import { DeferredEntity } from './processing/types';
 
 export interface LocationService {
   createLocation(
@@ -39,8 +40,8 @@ export interface CatalogProcessingEngine {
 }
 
 export type EntityProviderMutation =
-  | { type: 'full'; entities: Entity[] }
-  | { type: 'delta'; added: Entity[]; removed: Entity[] };
+  | { type: 'full'; entities: DeferredEntity[] }
+  | { type: 'delta'; added: DeferredEntity[]; removed: DeferredEntity[] };
 
 export interface EntityProviderConnection {
   applyMutation(mutation: EntityProviderMutation): Promise<void>;