From 61d4efe9787afbe9beb0b53f0808dad45a0f6e18 Mon Sep 17 00:00:00 2001
From: Damon Kaswell <damon.kaswell1@hp.com>
Date: Mon, 28 Nov 2022 07:50:31 -0800
Subject: [PATCH] Update changeset

Signed-off-by: Damon Kaswell <damon.kaswell1@hp.com>
---
 .changeset/sharp-shoes-bathe.md               |  5 +++
 .../IncrementalIngestionDatabaseManager.ts    | 30 ++++++++++++-----
 .../src/engine/IncrementalIngestionEngine.ts  | 33 ++++++++++++++++---
 3 files changed, 54 insertions(+), 14 deletions(-)
 create mode 100644 .changeset/sharp-shoes-bathe.md

diff --git a/.changeset/sharp-shoes-bathe.md b/.changeset/sharp-shoes-bathe.md
new file mode 100644
index 0000000000..7c44559e2c
--- /dev/null
+++ b/.changeset/sharp-shoes-bathe.md
@@ -0,0 +1,5 @@
+---
+'@backstage/plugin-catalog-backend-module-incremental-ingestion': minor
+---
+
+Make incremental providers more resilient to failures
diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts
index 3c91b24e2c..c6d5fe795b 100644
--- a/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts
+++ b/plugins/catalog-backend-module-incremental-ingestion/src/database/IncrementalIngestionDatabaseManager.ts
@@ -287,6 +287,15 @@ export class IncrementalIngestionDatabaseManager {
    */
   async computeRemoved(provider: string, ingestionId: string) {
     return await this.client.transaction(async tx => {
+      const rows = await tx('final_entities')
+        .count({ total: '*' })
+        .join('search', 'search.entity_id', 'final_entities.entity_id')
+        .where(
+          'search.key',
+          `metadata.annotations.${INCREMENTAL_ENTITY_PROVIDER_ANNOTATION}`,
+        )
+        .andWhere('search.value', provider);
+      const total = rows.reduce((acc, cur) => acc + (cur.total as number), 0);
       const removed: { entity: string; ref: string }[] = await tx(
         'final_entities',
       )
@@ -302,23 +311,26 @@ export class IncrementalIngestionDatabaseManager {
         .join('search', 'search.entity_id', 'final_entities.entity_id')
         .whereNotIn(
           'entity_ref',
-          tx('ingestion_marks')
+          tx('ingestion.ingestion_marks')
             .join(
-              'ingestion_mark_entities',
-              'ingestion_marks.id',
-              'ingestion_mark_entities.ingestion_mark_id',
+              'ingestion.ingestion_mark_entities',
+              'ingestion.ingestion_marks.id',
+              'ingestion.ingestion_mark_entities.ingestion_mark_id',
             )
-            .select('ingestion_mark_entities.ref')
-            .where('ingestion_marks.ingestion_id', ingestionId),
+            .select('ingestion.ingestion_mark_entities.ref')
+            .where('ingestion.ingestion_marks.ingestion_id', ingestionId),
         )
         .andWhere(
           'search.key',
           `metadata.annotations.${INCREMENTAL_ENTITY_PROVIDER_ANNOTATION}`,
         )
         .andWhere('search.value', provider);
-      return removed.map(entity => {
-        return { entity: JSON.parse(entity.entity) };
-      });
+      return {
+        total,
+        removed: removed.map(entity => {
+          return { entity: JSON.parse(entity.entity) };
+        }),
+      };
     });
   }
 
diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts
index a742f63045..36b683fa9a 100644
--- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts
+++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts
@@ -27,6 +27,8 @@ 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,12 +281,33 @@ export class IncrementalIngestionEngine implements IterationEngine {
     const removed: DeferredEntity[] = [];
 
     if (done) {
-      removed.push(
-        ...(await this.manager.computeRemoved(
-          this.options.provider.getProviderName(),
-          id,
-        )),
+      this.options.logger.info(
+        `incremental-engine: Ingestion '${id}': Final page reached, calculating removed entities`,
       );
+      const result = await this.manager.computeRemoved(
+        this.options.provider.getProviderName(),
+        id,
+      );
+      const total = result.total + added.length;
+      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`,
+        );
+        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}`,
+          },
+        });
+      }
     }
 
     await this.options.connection.applyMutation({