diff --git a/plugins/catalog-backend/package.json b/plugins/catalog-backend/package.json index 0bf5a08dfd..f095f39785 100644 --- a/plugins/catalog-backend/package.json +++ b/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", diff --git a/plugins/catalog-backend/src/next/NextCatalogBuilder.ts b/plugins/catalog-backend/src/next/NextCatalogBuilder.ts index fb12077508..fe7df180ee 100644 --- a/plugins/catalog-backend/src/next/NextCatalogBuilder.ts +++ b/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({ diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts index 5103c25dbc..690dcc1b7b 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts +++ b/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('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('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, + ); }); }); diff --git a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts b/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts index 177bf979ab..926423c848 100644 --- a/plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts +++ b/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 { + 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 {