Added a mechanism to perform out-of-sequence deltas

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

plugins/catalog-backend-module-incremental-ingestion/package.json

@@ -40,6 +40,7 @@
     "@backstage/catalog-model": "workspace:^",
     "@backstage/config": "workspace:^",
     "@backstage/errors": "workspace:^",
+    "@backstage/plugin-events-node": "workspace:^",
     "@backstage/plugin-catalog-backend": "workspace:^",
     "@backstage/plugin-catalog-node": "workspace:^",
     "@backstage/plugin-permission-common": "workspace:^",

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

@@ -21,14 +21,18 @@ import { performance } from 'perf_hooks';
 import { Duration, DurationObjectUnits } from 'luxon';
 import { v4 } from 'uuid';
 import { stringifyError } from '@backstage/errors';
+import { EventParams, EventSubscriber } from '@backstage/plugin-events-node';
 
-export class IncrementalIngestionEngine implements IterationEngine {
+export class IncrementalIngestionEngine<TInput>
+  implements IterationEngine, EventSubscriber
+{
   private readonly restLength: Duration;
   private readonly backoff: DurationObjectUnits[];
+  private readonly providerEventTopic: string;
 
   private manager: IncrementalIngestionDatabaseManager;
 
-  constructor(private options: IterationEngineOptions) {
+  constructor(private options: IterationEngineOptions<TInput>) {
     this.manager = options.manager;
     this.restLength = Duration.fromObject(options.restLength);
     this.backoff = options.backoff ?? [
@@ -37,6 +41,7 @@ export class IncrementalIngestionEngine implements IterationEngine {
       { minutes: 30 },
       { hours: 3 },
     ];
+    this.providerEventTopic = `${options.provider.getProviderName()}-delta`;
   }
 
   async taskFn(signal: AbortSignal) {
@@ -326,4 +331,42 @@ export class IncrementalIngestionEngine implements IterationEngine {
       removed,
     });
   }
+
+  async onEvent(params: EventParams): Promise<void> {
+    const { topic, eventPayload } = params;
+    if (topic !== this.providerEventTopic) {
+      return;
+    }
+
+    const { logger, provider, connection } = this.options;
+    logger.info(
+      `incremental-engine: Received ${this.providerEventTopic} event`,
+    );
+
+    const payload = eventPayload as TInput;
+
+    if (!provider.deltaMapper) {
+      return;
+    }
+
+    const update = provider.deltaMapper(payload);
+
+    if (update.delta) {
+      await connection.applyMutation({
+        type: 'delta',
+        ...update.delta,
+      });
+      logger.info(
+        `incremental-engine: Processed ${this.providerEventTopic} event`,
+      );
+    } else {
+      logger.info(
+        `incremental-engine: Rejected ${this.providerEventTopic} event - empty or invalid`,
+      );
+    }
+  }
+
+  supportsEventTopics(): string[] {
+    return [this.providerEventTopic];
+  }
 }

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

@@ -28,7 +28,7 @@ import { Duration } from 'luxon';
 import { IncrementalIngestionDatabaseManager } from '../database/IncrementalIngestionDatabaseManager';
 import { applyDatabaseMigrations } from '../database/migrations';
 import { IncrementalIngestionEngine } from '../engine/IncrementalIngestionEngine';
-import { createIncrementalProviderRouter } from '../router/routes';
+import { IncrementalProviderRouter } from '../router/routes';
 import {
   IncrementalEntityProvider,
   IncrementalEntityProviderOptions,
@@ -72,14 +72,14 @@ export class WrapperProviders {
   }
 
   async adminRouter(): Promise<express.Router> {
-    return createIncrementalProviderRouter(
+    return await new IncrementalProviderRouter(
       new IncrementalIngestionDatabaseManager({ client: this.options.client }),
       loggerToWinstonLogger(this.options.logger),
-    );
+    ).createRouter();
   }
 
-  private async startProvider(
-    provider: IncrementalEntityProvider<unknown, unknown>,
+  private async startProvider<TCursor, TContext, TInput>(
+    provider: IncrementalEntityProvider<TCursor, TContext, TInput>,
     providerOptions: IncrementalEntityProviderOptions,
     connection: EntityProviderConnection,
   ) {

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

@@ -38,7 +38,7 @@ export const incrementalIngestionEntityProviderCatalogModule =
       env,
       options: {
         providers: Array<{
-          provider: IncrementalEntityProvider<unknown, unknown>;
+          provider: IncrementalEntityProvider<unknown, unknown, unknown>;
           options: IncrementalEntityProviderOptions;
         }>;
       },

plugins/catalog-backend-module-incremental-ingestion/src/router/routes.ts

@@ -15,218 +15,269 @@
  */
 
 import { errorHandler } from '@backstage/backend-common';
+import { stringifyError } from '@backstage/errors';
+import { EventBroker, EventPublisher } from '@backstage/plugin-events-node';
 import express from 'express';
 import Router from 'express-promise-router';
 import { Logger } from 'winston';
 import { IncrementalIngestionDatabaseManager } from '../database/IncrementalIngestionDatabaseManager';
 import { PROVIDER_BASE_PATH, PROVIDER_CLEANUP, PROVIDER_HEALTH } from './paths';
 
-export const createIncrementalProviderRouter = async (
-  manager: IncrementalIngestionDatabaseManager,
-  logger: Logger,
-) => {
-  const router = Router();
-  router.use(express.json());
+export class IncrementalProviderRouter implements EventPublisher {
+  private manager: IncrementalIngestionDatabaseManager;
+  private logger: Logger;
+  private eventBroker: EventBroker | undefined;
 
-  // Get the overall health of all incremental providers
-  router.get(PROVIDER_HEALTH, async (_, res) => {
-    const records = await manager.healthcheck();
-    const providers = records.map(record => record.provider_name);
-    const duplicates = [
-      ...new Set(providers.filter((e, i, a) => a.indexOf(e) !== i)),
-    ];
+  constructor(manager: IncrementalIngestionDatabaseManager, logger: Logger) {
+    this.manager = manager;
+    this.logger = logger;
+  }
 
-    if (duplicates.length > 0) {
-      res.json({ healthy: false, duplicateIngestions: duplicates });
-    } else {
-      res.json({ healthy: true });
-    }
-  });
+  async setEventBroker(eventBroker: EventBroker): Promise<void> {
+    this.eventBroker = eventBroker;
+  }
 
-  // Clean up and pause all providers
-  router.post(PROVIDER_CLEANUP, async (_, res) => {
-    const result = await manager.cleanupProviders();
-    res.json(result);
-  });
+  async createRouter() {
+    const router = Router();
+    router.use(express.json());
 
-  // Get basic status of the provider
-  router.get(PROVIDER_BASE_PATH, async (req, res) => {
-    const { provider } = req.params;
-    const record = await manager.getCurrentIngestionRecord(provider);
-    if (record) {
-      res.json({
-        success: true,
-        status: {
-          current_action: record.status,
-          next_action_at: new Date(record.next_action_at),
-        },
-        last_error: record.last_error,
-      });
-    } else {
-      const providers: string[] = await manager.listProviders();
-      if (providers.includes(provider)) {
+    // Get the overall health of all incremental providers
+    router.get(PROVIDER_HEALTH, async (_, res) => {
+      const records = await this.manager.healthcheck();
+      const providers = records.map(record => record.provider_name);
+      const duplicates = [
+        ...new Set(providers.filter((e, i, a) => a.indexOf(e) !== i)),
+      ];
+
+      if (duplicates.length > 0) {
+        res.json({ healthy: false, duplicateIngestions: duplicates });
+      } else {
+        res.json({ healthy: true });
+      }
+    });
+
+    // Clean up and pause all providers
+    router.post(PROVIDER_CLEANUP, async (_, res) => {
+      const result = await this.manager.cleanupProviders();
+      res.json(result);
+    });
+
+    // Get basic status of the provider
+    router.get(PROVIDER_BASE_PATH, async (req, res) => {
+      const { provider } = req.params;
+      const record = await this.manager.getCurrentIngestionRecord(provider);
+      if (record) {
         res.json({
           success: true,
           status: {
-            current_action: 'rest complete, waiting to start',
+            current_action: record.status,
+            next_action_at: new Date(record.next_action_at),
           },
+          last_error: record.last_error,
         });
       } else {
-        logger.error(
-          `${provider} - No ingestion record found in the database!`,
-        );
-        res.status(404).json({
-          success: false,
-          status: {},
-          last_error: `Provider '${provider}' not found`,
-        });
+        const providers: string[] = await this.manager.listProviders();
+        if (providers.includes(provider)) {
+          res.json({
+            success: true,
+            status: {
+              current_action: 'rest complete, waiting to start',
+            },
+          });
+        } else {
+          this.logger.error(
+            `${provider} - No ingestion record found in the database!`,
+          );
+          res.status(404).json({
+            success: false,
+            status: {},
+            last_error: `Provider '${provider}' not found`,
+          });
+        }
       }
-    }
-  });
-
-  // Trigger the provider's next action
-  router.post(`${PROVIDER_BASE_PATH}/trigger`, async (req, res) => {
-    const { provider } = req.params;
-    const record = await manager.getCurrentIngestionRecord(provider);
-    if (record) {
-      await manager.triggerNextProviderAction(provider);
-      res.json({
-        success: true,
-        message: `${provider}: Next action triggered.`,
-      });
-    } else {
-      const providers: string[] = await manager.listProviders();
-      if (providers.includes(provider)) {
-        logger.debug(`${provider} - Ingestion record found`);
-        res.json({
-          success: true,
-          message: 'Unable to trigger next action (provider is restarting)',
-        });
-      } else {
-        res.status(404).json({
-          success: false,
-          message: `Provider '${provider}' not found`,
-        });
-      }
-    }
-  });
-
-  // Start a brand-new ingestion cycle for the provider.
-  // (Cancel's the current run if active, or marks it complete if resting)
-  router.post(`${PROVIDER_BASE_PATH}/start`, async (req, res) => {
-    const { provider } = req.params;
-
-    const record = await manager.getCurrentIngestionRecord(provider);
-    if (record) {
-      const ingestionId = record.id;
-      if (record.status === 'resting') {
-        await manager.setProviderComplete(ingestionId);
-      } else {
-        await manager.setProviderCanceling(ingestionId);
-      }
-      res.json({
-        success: true,
-        message: `${provider}: Next cycle triggered.`,
-      });
-    } else {
-      const providers: string[] = await manager.listProviders();
-      if (providers.includes(provider)) {
-        logger.debug(`${provider} - Ingestion record found`);
-        res.json({
-          success: true,
-          message: 'Provider is already restarting',
-        });
-      } else {
-        res.status(404).json({
-          success: false,
-          message: `Provider '${provider}' not found`,
-        });
-      }
-    }
-  });
-
-  // Stop the provider and pause it for 24 hours
-  router.post(`${PROVIDER_BASE_PATH}/cancel`, async (req, res) => {
-    const { provider } = req.params;
-    const record = await manager.getCurrentIngestionRecord(provider);
-    if (record) {
-      const next_action_at = new Date();
-      next_action_at.setTime(next_action_at.getTime() + 24 * 60 * 60 * 1000);
-      await manager.updateByName(provider, {
-        next_action: 'nothing (done)',
-        ingestion_completed_at: new Date(),
-        next_action_at,
-        status: 'resting',
-      });
-      res.json({
-        success: true,
-        message: `${provider}: Current ingestion canceled.`,
-      });
-    } else {
-      const providers: string[] = await manager.listProviders();
-      if (providers.includes(provider)) {
-        logger.debug(`${provider} - Ingestion record found`);
-        res.json({
-          success: true,
-          message: 'Provider is currently restarting, please wait.',
-        });
-      } else {
-        res.status(404).json({
-          success: false,
-          message: `Provider '${provider}' not found`,
-        });
-      }
-    }
-  });
-
-  // Wipe out all ingestion records for the provider and pause for 24 hours
-  router.delete(PROVIDER_BASE_PATH, async (req, res) => {
-    const { provider } = req.params;
-    const result = await manager.purgeAndResetProvider(provider);
-    res.json(result);
-  });
-
-  // Get the ingestion marks for the current cycle
-  router.get(`${PROVIDER_BASE_PATH}/marks`, async (req, res) => {
-    const { provider } = req.params;
-    const record = await manager.getCurrentIngestionRecord(provider);
-    if (record) {
-      const id = record.id;
-      const records = await manager.getAllMarks(id);
-      res.json({ success: true, records });
-    } else {
-      const providers: string[] = await manager.listProviders();
-      if (providers.includes(provider)) {
-        logger.debug(`${provider} - Ingestion record found`);
-        res.json({
-          success: true,
-          message: 'No records yet (provider is restarting)',
-        });
-      } else {
-        logger.error(
-          `${provider} - No ingestion record found in the database!`,
-        );
-        res.status(404).json({
-          success: false,
-          status: {},
-          last_error: `Provider '${provider}' not found`,
-        });
-      }
-    }
-  });
-
-  router.delete(`${PROVIDER_BASE_PATH}/marks`, async (req, res) => {
-    const { provider } = req.params;
-    const deletions = await manager.clearFinishedIngestions(provider);
-
-    res.json({
-      success: true,
-      message: `Expired marks for provider '${provider}' removed.`,
-      deletions,
     });
-  });
 
-  router.use(errorHandler());
+    // Trigger the provider's next action
+    router.post(`${PROVIDER_BASE_PATH}/trigger`, async (req, res) => {
+      const { provider } = req.params;
+      const record = await this.manager.getCurrentIngestionRecord(provider);
+      if (record) {
+        await this.manager.triggerNextProviderAction(provider);
+        res.json({
+          success: true,
+          message: `${provider}: Next action triggered.`,
+        });
+      } else {
+        const providers: string[] = await this.manager.listProviders();
+        if (providers.includes(provider)) {
+          this.logger.debug(`${provider} - Ingestion record found`);
+          res.json({
+            success: true,
+            message: 'Unable to trigger next action (provider is restarting)',
+          });
+        } else {
+          res.status(404).json({
+            success: false,
+            message: `Provider '${provider}' not found`,
+          });
+        }
+      }
+    });
 
-  return router;
-};
+    // Start a brand-new ingestion cycle for the provider.
+    // (Cancel's the current run if active, or marks it complete if resting)
+    router.post(`${PROVIDER_BASE_PATH}/start`, async (req, res) => {
+      const { provider } = req.params;
+
+      const record = await this.manager.getCurrentIngestionRecord(provider);
+      if (record) {
+        const ingestionId = record.id;
+        if (record.status === 'resting') {
+          await this.manager.setProviderComplete(ingestionId);
+        } else {
+          await this.manager.setProviderCanceling(ingestionId);
+        }
+        res.json({
+          success: true,
+          message: `${provider}: Next cycle triggered.`,
+        });
+      } else {
+        const providers: string[] = await this.manager.listProviders();
+        if (providers.includes(provider)) {
+          this.logger.debug(`${provider} - Ingestion record found`);
+          res.json({
+            success: true,
+            message: 'Provider is already restarting',
+          });
+        } else {
+          res.status(404).json({
+            success: false,
+            message: `Provider '${provider}' not found`,
+          });
+        }
+      }
+    });
+
+    // Stop the provider and pause it for 24 hours
+    router.post(`${PROVIDER_BASE_PATH}/cancel`, async (req, res) => {
+      const { provider } = req.params;
+      const record = await this.manager.getCurrentIngestionRecord(provider);
+      if (record) {
+        const next_action_at = new Date();
+        next_action_at.setTime(next_action_at.getTime() + 24 * 60 * 60 * 1000);
+        await this.manager.updateByName(provider, {
+          next_action: 'nothing (done)',
+          ingestion_completed_at: new Date(),
+          next_action_at,
+          status: 'resting',
+        });
+        res.json({
+          success: true,
+          message: `${provider}: Current ingestion canceled.`,
+        });
+      } else {
+        const providers: string[] = await this.manager.listProviders();
+        if (providers.includes(provider)) {
+          this.logger.debug(`${provider} - Ingestion record found`);
+          res.json({
+            success: true,
+            message: 'Provider is currently restarting, please wait.',
+          });
+        } else {
+          res.status(404).json({
+            success: false,
+            message: `Provider '${provider}' not found`,
+          });
+        }
+      }
+    });
+
+    // Wipe out all ingestion records for the provider and pause for 24 hours
+    router.delete(PROVIDER_BASE_PATH, async (req, res) => {
+      const { provider } = req.params;
+      const result = await this.manager.purgeAndResetProvider(provider);
+      res.json(result);
+    });
+
+    // Get the ingestion marks for the current cycle
+    router.get(`${PROVIDER_BASE_PATH}/marks`, async (req, res) => {
+      const { provider } = req.params;
+      const record = await this.manager.getCurrentIngestionRecord(provider);
+      if (record) {
+        const id = record.id;
+        const records = await this.manager.getAllMarks(id);
+        res.json({ success: true, records });
+      } else {
+        const providers: string[] = await this.manager.listProviders();
+        if (providers.includes(provider)) {
+          this.logger.debug(`${provider} - Ingestion record found`);
+          res.json({
+            success: true,
+            message: 'No records yet (provider is restarting)',
+          });
+        } else {
+          this.logger.error(
+            `${provider} - No ingestion record found in the database!`,
+          );
+          res.status(404).json({
+            success: false,
+            status: {},
+            last_error: `Provider '${provider}' not found`,
+          });
+        }
+      }
+    });
+
+    router.delete(`${PROVIDER_BASE_PATH}/marks`, async (req, res) => {
+      const { provider } = req.params;
+      const deletions = await this.manager.clearFinishedIngestions(provider);
+
+      res.json({
+        success: true,
+        message: `Expired marks for provider '${provider}' removed.`,
+        deletions,
+      });
+    });
+
+    router.post(`${PROVIDER_BASE_PATH}/delta`, async (req, res) => {
+      const { provider } = req.params;
+
+      const topic = `${provider}-push`;
+
+      const eventPayload = req.body;
+
+      if (!this.eventBroker) {
+        res.status(500).json({
+          success: false,
+          provider,
+          message: `The payload could not be processed!`,
+        });
+        throw new Error('Event broker not initialized!');
+      }
+
+      try {
+        await this.eventBroker.publish({
+          topic,
+          eventPayload,
+        });
+        res.json({
+          success: true,
+          provider,
+          message: 'Payload submitted.',
+        });
+      } catch (e) {
+        res.status(500).json({
+          success: false,
+          provider,
+          message: `There was an error submitting the payload: ${stringifyError(
+            e,
+          )}`,
+        });
+      }
+    });
+
+    router.use(errorHandler());
+
+    return router;
+  }
+}

plugins/catalog-backend-module-incremental-ingestion/src/service/IncrementalCatalogBuilder.ts

@@ -24,7 +24,7 @@ import { Knex } from 'knex';
 import { IncrementalIngestionEngine } from '../engine/IncrementalIngestionEngine';
 import { applyDatabaseMigrations } from '../database/migrations';
 import { IncrementalIngestionDatabaseManager } from '../database/IncrementalIngestionDatabaseManager';
-import { createIncrementalProviderRouter } from '../router/routes';
+import { IncrementalProviderRouter } from '../router/routes';
 import { Deferred } from '../util';
 
 /** @public */
@@ -60,16 +60,16 @@ export class IncrementalCatalogBuilder {
       router: 'IncrementalProviderAdmin',
     });
 
-    const incrementalAdminRouter = await createIncrementalProviderRouter(
+    const incrementalAdminRouter = await new IncrementalProviderRouter(
       this.manager,
       routerLogger,
-    );
+    ).createRouter();
 
     return { incrementalAdminRouter };
   }
 
-  addIncrementalEntityProvider<TCursor, TContext>(
-    provider: IncrementalEntityProvider<TCursor, TContext>,
+  addIncrementalEntityProvider<TCursor, TContext, TInput>(
+    provider: IncrementalEntityProvider<TCursor, TContext, TInput>,
     options: IncrementalEntityProviderOptions,
   ) {
     const { burstInterval, burstLength, restLength } = options;

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

@@ -46,7 +46,7 @@ import { IncrementalIngestionDatabaseManager } from './database/IncrementalInges
  *
  * @public
  */
-export interface IncrementalEntityProvider<TCursor, TContext> {
+export interface IncrementalEntityProvider<TCursor, TContext, TInput = null> {
   /**
    * This name must be unique between all of the entity providers
    * operating in the catalog.
@@ -75,6 +75,14 @@ export interface IncrementalEntityProvider<TCursor, TContext> {
    * @param burst - a function which performs a series of iterations
    */
   around(burst: (context: TContext) => Promise<void>): Promise<void>;
+
+  /**
+   * If present, this method maps incoming payloads to apply updates
+   * outside of the incremental ingestion schedule.
+   */
+  deltaMapper?: (payload: TInput) => {
+    delta: { added: DeferredEntity[]; removed: DeferredEntity[] } | undefined;
+  };
 }
 
 /**
@@ -154,11 +162,11 @@ export interface IterationEngine {
   taskFn: TaskFunction;
 }
 
-export interface IterationEngineOptions {
+export interface IterationEngineOptions<TInput> {
   logger: Logger;
   connection: EntityProviderConnection;
   manager: IncrementalIngestionDatabaseManager;
-  provider: IncrementalEntityProvider<unknown, unknown>;
+  provider: IncrementalEntityProvider<unknown, unknown, TInput>;
   restLength: DurationObjectUnits;
   ready: Promise<void>;
   backoff?: IncrementalEntityProviderOptions['backoff'];