Implement configurable threshold and empty collection check

Signed-off-by: Damon Kaswell <damon.kaswell1@hp.com>

plugins/catalog-backend-module-incremental-ingestion/api-report.md

@@ -62,6 +62,8 @@ export interface IncrementalEntityProviderOptions {
   backoff?: DurationObjectUnits[];
   burstInterval: DurationObjectUnits;
   burstLength: DurationObjectUnits;
+  rejectEmptyEntityCollections?: boolean;
+  removalThreshold?: number;
   restLength: DurationObjectUnits;
 }
 

plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts

@@ -298,11 +298,24 @@ export class IncrementalIngestionDatabaseManager {
    * @param ingestionId - string
    * @returns All entities to remove for this burst.
    */
-  async computeRemoved(provider: string) {
+  async computeRemoved(provider: string, providerId: string) {
     const previousIngestion = await this.getPreviousIngestionRecord(provider);
     return await this.client.transaction(async tx => {
+      const count = await tx('ingestion_mark_entities')
+        .count({ total: 'ingestion_mark_entities.ref' })
+        .join(
+          'ingestion_marks',
+          'ingestion_marks.id',
+          'ingestion_mark_entities.ingestion_mark_id',
+        )
+        .join('ingestions', 'ingestions.id', 'ingestion_marks.ingestion_id')
+        .where('ingestions.id', providerId);
+
+      const total = count.reduce((acc, cur) => acc + (cur.total as number), 0);
+
+      const removed: { entityRef: string }[] = [];
       if (previousIngestion) {
-        const rows: { ref: string }[] = await tx('ingestion_mark_entities')
+        const stale: { ref: string }[] = await tx('ingestion_mark_entities')
           .select('ingestion_mark_entities.ref')
           .join(
             'ingestion_marks',
@@ -312,14 +325,14 @@ export class IncrementalIngestionDatabaseManager {
           .join('ingestions', 'ingestions.id', 'ingestion_marks.ingestion_id')
           .where('ingestions.id', previousIngestion.id);
 
-        const removed: { entityRef: string }[] = rows.map(e => {
-          return { entityRef: e.ref };
-        });
-        const total = rows.length ?? 0;
-        return { removed, total };
+        removed.push(
+          ...stale.map(e => {
+            return { entityRef: e.ref };
+          }),
+        );
       }
 
-      return { removed: [], total: 0 };
+      return { total, removed };
     });
   }
 

plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts

@@ -22,8 +22,6 @@ import { Duration, DurationObjectUnits } from 'luxon';
 import { v4 } from 'uuid';
 import { stringifyError } from '@backstage/errors';
 
-const REMOVAL_THRESHOLD = 5;
-
 export class IncrementalIngestionEngine implements IterationEngine {
   private readonly restLength: Duration;
   private readonly backoff: DurationObjectUnits[];
@@ -279,26 +277,46 @@ export class IncrementalIngestionEngine implements IterationEngine {
       );
       const result = await this.manager.computeRemoved(
         this.options.provider.getProviderName(),
+        id,
       );
+
       const { total } = result;
-      const percentRemoved =
-        total > 0 ? (result.removed.length / total) * 100 : 0;
-      if (percentRemoved <= REMOVAL_THRESHOLD) {
-        this.options.logger.info(
-          `incremental-engine: Ingestion '${id}': Removing ${result.removed.length} entities that have no matching assets`,
-        );
+
+      let doRemoval = true;
+      if (this.options.rejectEmptyEntityCollections) {
+        if (total === 0) {
+          this.options.logger.error(
+            `incremental-engine: Ingestion '${id}': Rejecting empty entity collection!`,
+          );
+          doRemoval = false;
+        }
+      }
+
+      if (this.options.removalThreshold) {
+        // If the total entities upserted in this ingestion is 0, then
+        // 100% of entities are stale and marked for removal.
+        const percentRemoved =
+          total > 0 ? (result.removed.length / total) * 100 : 100;
+        if (percentRemoved <= this.options.removalThreshold) {
+          this.options.logger.info(
+            `incremental-engine: Ingestion '${id}': Removing ${result.removed.length} entities that have no matching assets`,
+          );
+        } else {
+          const notice = `Attempted to remove ${percentRemoved}% of matching entities!`;
+          this.options.logger.error(
+            `incremental-engine: Ingestion '${id}': ${notice}`,
+          );
+          await this.manager.updateIngestionRecordById({
+            ingestionId: id,
+            update: {
+              last_error: `REMOVAL_THRESHOLD exceeded on ingestion mark ${markId}: ${notice}`,
+            },
+          });
+          doRemoval = false;
+        }
+      }
+      if (doRemoval) {
         removed.push(...result.removed);
-      } else {
-        const notice = `Attempted to remove ${percentRemoved}% of ${total} matching entities!`;
-        this.options.logger.error(
-          `incremental-engine: Ingestion '${id}': ${notice}`,
-        );
-        await this.manager.updateIngestionRecordById({
-          ingestionId: id,
-          update: {
-            last_error: `REMOVAL_THRESHOLD exceeded on ingestion mark ${markId}: ${notice}`,
-          },
-        });
       }
     }
 

plugins/catalog-backend-module-incremental-ingestion/src/types.ts

@@ -122,6 +122,22 @@ export interface IncrementalEntityProviderOptions {
    * `[{ minutes: 1}, { minutes: 5}, {minutes: 30 }, { hours: 3 }]`
    */
   backoff?: DurationObjectUnits[];
+
+  /**
+   * If an error occurs at a data source that results in a large
+   * number of assets being inadvertently removed, it will result in
+   * Backstage removing all associated entities. To avoid that, set
+   * a percentage of entities past which removal will be disallowed.
+   */
+  removalThreshold?: number;
+
+  /**
+   * Similar to the removalThreshold, this option prevents removals
+   * in circumstances where a data source has improperly returned 0
+   * assets. If set to `true`, Backstage will reject removals when
+   * that happens.
+   */
+  rejectEmptyEntityCollections?: boolean;
 }
 
 /** @public */
@@ -146,4 +162,6 @@ export interface IterationEngineOptions {
   restLength: DurationObjectUnits;
   ready: Promise<void>;
   backoff?: IncrementalEntityProviderOptions['backoff'];
+  removalThreshold?: number;
+  rejectEmptyEntityCollections?: boolean;
 }