Catalog: Add refresh spread functionality

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

plugins/catalog-backend/package.json

@@ -73,7 +73,8 @@
     "msw": "^0.29.0",
     "sqlite3": "^5.0.1",
     "supertest": "^6.1.3",
-    "wait-for-expect": "^3.0.2"
+    "wait-for-expect": "^3.0.2",
+    "luxon": "^2.0.2"
   },
   "files": [
     "dist",

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

@@ -113,6 +113,7 @@ export class NextCatalogBuilder {
   private processorsReplace: boolean;
   private parser: CatalogProcessorParser | undefined;
   private refreshIntervalSeconds = 100;
+  private refreshSpreadSeconds = { min: 10, max: 60 };
 
   constructor(env: CatalogEnvironment) {
     this.env = env;
@@ -152,6 +153,15 @@ export class NextCatalogBuilder {
     return this;
   }
 
+  /**
+   * Refresh spread configures configures the minimum and maximum number of seconds
+   * to wait between refreshes in addition to the configured refresh interval.
+   */
+  setRefreshSpreadSeconds(min: number, max: number): NextCatalogBuilder {
+    this.refreshSpreadSeconds = { min, max };
+    return this;
+  }
+
   /**
    * Sets what policies to use for validation of entities between the pre-
    * processing and post-processing stages. All such policies must pass for the
@@ -286,6 +296,7 @@ export class NextCatalogBuilder {
       database: dbClient,
       logger,
       refreshIntervalSeconds: this.refreshIntervalSeconds,
+      refreshSpreadSeconds: this.refreshSpreadSeconds,
     });
     const integrations = ScmIntegrations.fromConfig(config);
     const orchestrator = new DefaultCatalogProcessingOrchestrator({

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

@@ -20,7 +20,11 @@ import { Entity, stringifyEntityRef } from '@backstage/catalog-model';
 import { JsonObject } from '@backstage/config';
 import { Knex } from 'knex';
 import * as uuid from 'uuid';
+<<<<<<< HEAD
 import { Logger } from 'winston';
+=======
+import { DateTime } from 'luxon';
+>>>>>>> d31a82869 (Catalog: Add refresh spread functionality)
 import { DatabaseManager } from './DatabaseManager';
 import { DefaultProcessingDatabase } from './DefaultProcessingDatabase';
 import {
@@ -47,6 +51,7 @@ describe('Default Processing Database', () => {
         database: knex,
         logger,
         refreshIntervalSeconds: 100,
+        refreshSpreadSeconds: { min: 10, max: 60 },
       }),
     };
   }
@@ -960,5 +965,44 @@ describe('Default Processing Database', () => {
       },
       60_000,
     );
+
+    it.each(databases.eachSupportedId())(
+      'should update the next_refresh interval with a timestamp that includes refresh spread, %p',
+      async databaseId => {
+        const { knex, db } = await createDatabase(databaseId);
+        const entity = JSON.stringify({
+          kind: 'Location',
+          apiVersion: '1.0.0',
+          metadata: {
+            name: 'xyz',
+          },
+        } as Entity);
+        await knex<DbRefreshStateRow>('refresh_state').insert({
+          entity_id: '2',
+          entity_ref: 'location:default/new-root',
+          unprocessed_entity: entity,
+          errors: '[]',
+          next_update_at: '2019-01-01 23:00:00',
+          last_discovery_at: '2021-04-01 13:37:00',
+        });
+        await db.transaction(async tx => {
+          // Result does not include the updated timestamp
+          await db.getProcessableEntities(tx, {
+            processBatchSize: 1,
+          });
+        });
+        const now = DateTime.local();
+        const result = await knex<DbRefreshStateRow>('refresh_state')
+          .where('entity_ref', 'location:default/new-root')
+          .select();
+        const nextUpdate = DateTime.fromSQL(result[0].next_update_at, {
+          zone: 'utc',
+        });
+        expect(nextUpdate.diff(now, 'seconds').seconds).toBeGreaterThanOrEqual(
+          110,
+        );
+      },
+      60_000,
+    );
   });
 });

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

@@ -49,6 +49,10 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
       database: Knex;
       logger: Logger;
       refreshIntervalSeconds: number;
+      refreshSpreadSeconds: {
+        min: number;
+        max: number;
+      };
     },
   ) {}
 
@@ -447,6 +451,17 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
     }
   }
 
+  // Returns the next update timestamp by combining refresh interval and refresh spread
+  private getNextUpdateAt(tx: Knex.Transaction): Knex.MaybeRawColumn<string> {
+    const { min, max } = this.options.refreshSpreadSeconds;
+    const refreshSpread = Math.floor(Math.random() * (max - min + 1)) + min;
+    const nextUpdate = this.options.refreshIntervalSeconds + refreshSpread;
+
+    return tx.client.config.client === 'sqlite3'
+      ? tx.raw(`datetime('now', ?)`, [`${nextUpdate} seconds`])
+      : tx.raw(`now() + interval '${Number(nextUpdate)} seconds'`);
+  }
+
   async getProcessableEntities(
     txOpaque: Transaction,
     request: { processBatchSize: number },
@@ -473,16 +488,7 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
         items.map(i => i.entity_ref),
       )
       .update({
-        next_update_at:
-          tx.client.config.client === 'sqlite3'
-            ? tx.raw(`datetime('now', ?)`, [
-                `${this.options.refreshIntervalSeconds} seconds`,
-              ])
-            : tx.raw(
-                `now() + interval '${Number(
-                  this.options.refreshIntervalSeconds,
-                )} seconds'`,
-              ),
+        next_update_at: this.getNextUpdateAt(tx),
       });
 
     return {