From e2dd0952d16cd861669d18cc7d2885477017946f Mon Sep 17 00:00:00 2001 From: Chris Wyatt Cook Date: Mon, 23 Jun 2025 17:18:02 -0700 Subject: [PATCH 1/6] incremental burstLength check Signed-off-by: Chris Wyatt Cook --- .changeset/fancy-nails-call.md | 5 + .../engine/IncrementalIngestionEngine.test.ts | 173 ++++++++++++++++++ .../src/engine/IncrementalIngestionEngine.ts | 12 ++ .../src/types.ts | 1 + 4 files changed, 191 insertions(+) create mode 100644 .changeset/fancy-nails-call.md create mode 100644 plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.test.ts diff --git a/.changeset/fancy-nails-call.md b/.changeset/fancy-nails-call.md new file mode 100644 index 0000000000..97980029af --- /dev/null +++ b/.changeset/fancy-nails-call.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend-module-incremental-ingestion': patch +--- + +Fixed bug in IncrementalIngestionEngine by adding burstLength check when a burst completes diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.test.ts b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.test.ts new file mode 100644 index 0000000000..e6c7d67fd0 --- /dev/null +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.test.ts @@ -0,0 +1,173 @@ +/* + * Copyright 2025 The Backstage Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { IncrementalIngestionEngine } from './IncrementalIngestionEngine'; +import { IterationEngineOptions } from '../types'; +import { performance } from 'perf_hooks'; + +jest.setTimeout(60_000); + +describe('IncrementalIngestionEngine - Burst Length', () => { + const createMockProvider = () => ({ + getProviderName: jest.fn().mockReturnValue('test-provider'), + next: jest.fn(), + around: jest.fn(), + }); + + const createMockManager = () => + ({ + getLastMark: jest.fn().mockResolvedValue(null), + createMark: jest.fn().mockResolvedValue(undefined), + createMarkEntities: jest.fn().mockResolvedValue(undefined), + computeRemoved: jest.fn().mockResolvedValue({ total: 0, removed: [] }), + } as any); + + const createMockConnection = () => + ({ + applyMutation: jest.fn().mockResolvedValue(undefined), + refresh: jest.fn().mockResolvedValue(undefined), + } as any); + + const createMockLogger = () => + ({ + info: jest.fn(), + debug: jest.fn(), + error: jest.fn(), + warn: jest.fn(), + child: jest.fn().mockReturnThis(), + } as any); + + it('should respect burst length and stop burst when time limit exceeded', async () => { + const mockProvider = createMockProvider(); + const mockManager = createMockManager(); + const mockConnection = createMockConnection(); + const mockLogger = createMockLogger(); + + const options: IterationEngineOptions = { + provider: mockProvider, + manager: mockManager, + connection: mockConnection, + burstLength: { milliseconds: 100 }, // Short burst length for testing + restLength: { minutes: 1 }, + logger: mockLogger, + ready: Promise.resolve(), + }; + + const engine = new IncrementalIngestionEngine(options); + + let callCount = 0; + mockProvider.around.mockImplementation(async fn => { + await fn({}); + }); + + // Mock provider.next to return multiple batches that never complete + // Each call takes some time to simulate real processing + mockProvider.next.mockImplementation(async () => { + callCount++; + // Add a small delay to ensure we exceed burst length + await new Promise(resolve => setTimeout(resolve, 30)); + return { + done: false, // Never done - would continue forever without burst length + entities: [ + { + entity: { + kind: 'Component', + metadata: { name: `test-component-${callCount}` }, + }, + }, + ], + cursor: `cursor-${callCount}`, + }; + }); + + const signal = new AbortController().signal; + const start = performance.now(); + + const result = await engine.ingestOneBurst('test-ingestion', signal); + + const duration = performance.now() - start; + + // Verify that the burst was stopped due to time limit, not completion + expect(result).toBe(false); // Should return false since provider never returned done + expect(duration).toBeGreaterThanOrEqual(100); // Should have run for at least the burst length + expect(mockProvider.next).toHaveBeenCalledTimes(callCount); + expect(callCount).toBeGreaterThan(1); // Should have made multiple calls before stopping + + // Verify the correct log message was called + expect(mockLogger.info).toHaveBeenCalledWith( + expect.stringContaining('burst exceeded length of 100 milliseconds'), + ); + }); + + it('should complete burst normally when provider returns done before burst length', async () => { + const mockProvider = createMockProvider(); + const mockManager = createMockManager(); + const mockConnection = createMockConnection(); + const mockLogger = createMockLogger(); + + const options: IterationEngineOptions = { + provider: mockProvider, + manager: mockManager, + connection: mockConnection, + burstLength: { seconds: 10 }, // Long burst length + restLength: { minutes: 1 }, + logger: mockLogger, + ready: Promise.resolve(), + }; + + const engine = new IncrementalIngestionEngine(options); + + mockProvider.around.mockImplementation(async fn => { + await fn({}); + }); + + // Mock provider.next to return done after first call + mockProvider.next.mockResolvedValueOnce({ + done: true, + entities: [ + { + entity: { + kind: 'Component', + metadata: { name: 'test-component-1' }, + }, + }, + ], + cursor: 'final-cursor', + }); + + const signal = new AbortController().signal; + const result = await engine.ingestOneBurst('test-ingestion', signal); + + // Should return true when provider indicates done + expect(result).toBe(true); + expect(mockProvider.next).toHaveBeenCalledTimes(1); + + // Should NOT log the burst exceeded message + expect(mockLogger.info).not.toHaveBeenCalledWith( + expect.stringContaining('burst exceeded length of'), + ); + + // Should log burst initiation and completion + expect(mockLogger.info).toHaveBeenCalledWith( + "incremental-engine: Ingestion 'test-ingestion' burst initiated", + ); + expect(mockLogger.info).toHaveBeenCalledWith( + expect.stringContaining( + "incremental-engine: Ingestion 'test-ingestion' burst complete", + ), + ); + }); +}); 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 e005d29292..aeccaccf8d 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,7 @@ import { HumanDuration } from '@backstage/types'; export class IncrementalIngestionEngine implements IterationEngine { private readonly restLength: Duration; + private readonly burstLength: HumanDuration; private readonly backoff: HumanDuration[]; private readonly lastStarted: Gauge; private readonly lastCompleted: Gauge; @@ -38,6 +39,7 @@ export class IncrementalIngestionEngine implements IterationEngine { this.manager = options.manager; this.restLength = Duration.fromObject(options.restLength); + this.burstLength = options.burstLength; this.backoff = options.backoff ?? [ { minutes: 1 }, { minutes: 5 }, @@ -247,6 +249,16 @@ export class IncrementalIngestionEngine implements IterationEngine { }); if (signal.aborted || next.done) { break; + } else if ( + performance.now() - start > + Duration.fromObject(this.burstLength).as('milliseconds') + ) { + this.options.logger.info( + `incremental-engine: Ingestion '${id}' burst exceeded length of ${Duration.fromObject( + this.burstLength, + ).toHuman()}.`, + ); + break; } else { next = await this.options.provider.next(context, next.cursor); count++; diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts index 75aecf8fa5..04a73bd364 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/types.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/types.ts @@ -188,6 +188,7 @@ export interface IterationEngineOptions { manager: IncrementalIngestionDatabaseManager; provider: IncrementalEntityProvider; restLength: HumanDuration; + burstLength: HumanDuration; ready: Promise; backoff?: IncrementalEntityProviderOptions['backoff']; rejectRemovalsAbovePercentage?: number; From 2aa408f466af891b12ec301d80ad06ba8253b6cf Mon Sep 17 00:00:00 2001 From: Chris Wyatt Cook Date: Mon, 23 Jun 2025 18:00:40 -0700 Subject: [PATCH 2/6] update burstLength type to Duration Signed-off-by: Chris Wyatt Cook --- .../src/engine/IncrementalIngestionEngine.ts | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) 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 aeccaccf8d..4ecff02301 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts @@ -27,7 +27,7 @@ import { HumanDuration } from '@backstage/types'; export class IncrementalIngestionEngine implements IterationEngine { private readonly restLength: Duration; - private readonly burstLength: HumanDuration; + private readonly burstLength: Duration; private readonly backoff: HumanDuration[]; private readonly lastStarted: Gauge; private readonly lastCompleted: Gauge; @@ -39,7 +39,7 @@ export class IncrementalIngestionEngine implements IterationEngine { this.manager = options.manager; this.restLength = Duration.fromObject(options.restLength); - this.burstLength = options.burstLength; + this.burstLength = Duration.fromObject(options.burstLength); this.backoff = options.backoff ?? [ { minutes: 1 }, { minutes: 5 }, @@ -251,12 +251,10 @@ export class IncrementalIngestionEngine implements IterationEngine { break; } else if ( performance.now() - start > - Duration.fromObject(this.burstLength).as('milliseconds') + this.burstLength.as('milliseconds') ) { this.options.logger.info( - `incremental-engine: Ingestion '${id}' burst exceeded length of ${Duration.fromObject( - this.burstLength, - ).toHuman()}.`, + `incremental-engine: Ingestion '${id}' burst exceeded length of ${this.burstLength.toHuman()}.`, ); break; } else { From e784adc053966ab00b46170bca790c2502927aa7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Tue, 24 Jun 2025 13:52:43 +0200 Subject: [PATCH 3/6] Update .changeset/fancy-nails-call.md MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- .changeset/fancy-nails-call.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/fancy-nails-call.md b/.changeset/fancy-nails-call.md index 97980029af..556b358dfa 100644 --- a/.changeset/fancy-nails-call.md +++ b/.changeset/fancy-nails-call.md @@ -2,4 +2,4 @@ '@backstage/plugin-catalog-backend-module-incremental-ingestion': patch --- -Fixed bug in IncrementalIngestionEngine by adding burstLength check when a burst completes +Fixed bug in `IncrementalIngestionEngine` by adding `burstLength` check when a burst completes From d862a62521fcb0a0e09bf4812d7d15104aa0b570 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Tue, 24 Jun 2025 13:52:49 +0200 Subject: [PATCH 4/6] Update plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- .../src/engine/IncrementalIngestionEngine.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 4ecff02301..95510e78e4 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.ts @@ -254,7 +254,7 @@ export class IncrementalIngestionEngine implements IterationEngine { this.burstLength.as('milliseconds') ) { this.options.logger.info( - `incremental-engine: Ingestion '${id}' burst exceeded length of ${this.burstLength.toHuman()}.`, + `incremental-engine: Ingestion '${id}' burst ending after ${this.burstLength.toHuman()}.`, ); break; } else { From 3663b2763882e2ab118c1660cc68f7c4a17b81e2 Mon Sep 17 00:00:00 2001 From: Chris Wyatt Cook Date: Tue, 24 Jun 2025 06:47:15 -0700 Subject: [PATCH 5/6] update tests Signed-off-by: Chris Wyatt Cook --- .../src/engine/IncrementalIngestionEngine.test.ts | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.test.ts b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.test.ts index e6c7d67fd0..43f037218e 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.test.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.test.ts @@ -103,12 +103,13 @@ describe('IncrementalIngestionEngine - Burst Length', () => { // Verify that the burst was stopped due to time limit, not completion expect(result).toBe(false); // Should return false since provider never returned done expect(duration).toBeGreaterThanOrEqual(100); // Should have run for at least the burst length + expect(duration).toBeLessThan(200); // But not too much longer (allowing for timing variance) expect(mockProvider.next).toHaveBeenCalledTimes(callCount); expect(callCount).toBeGreaterThan(1); // Should have made multiple calls before stopping - // Verify the correct log message was called + // Verify that burst was terminated due to time limit (not normal completion) expect(mockLogger.info).toHaveBeenCalledWith( - expect.stringContaining('burst exceeded length of 100 milliseconds'), + expect.stringContaining('burst ending after'), ); }); @@ -157,7 +158,7 @@ describe('IncrementalIngestionEngine - Burst Length', () => { // Should NOT log the burst exceeded message expect(mockLogger.info).not.toHaveBeenCalledWith( - expect.stringContaining('burst exceeded length of'), + expect.stringContaining('burst ending after'), ); // Should log burst initiation and completion From bedaafdd4aa3b816df10db67343a5fd4888a13b6 Mon Sep 17 00:00:00 2001 From: Chris Wyatt Cook Date: Tue, 24 Jun 2025 07:30:07 -0700 Subject: [PATCH 6/6] remove logger.info tests from IncrementalIngestionEngine.test.ts Signed-off-by: Chris Wyatt Cook --- .../engine/IncrementalIngestionEngine.test.ts | 85 +++++++++++++------ 1 file changed, 61 insertions(+), 24 deletions(-) diff --git a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.test.ts b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.test.ts index 43f037218e..4bfe0666da 100644 --- a/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.test.ts +++ b/plugins/catalog-backend-module-incremental-ingestion/src/engine/IncrementalIngestionEngine.test.ts @@ -80,7 +80,7 @@ describe('IncrementalIngestionEngine - Burst Length', () => { // Add a small delay to ensure we exceed burst length await new Promise(resolve => setTimeout(resolve, 30)); return { - done: false, // Never done - would continue forever without burst length + done: false, entities: [ { entity: { @@ -101,16 +101,12 @@ describe('IncrementalIngestionEngine - Burst Length', () => { const duration = performance.now() - start; // Verify that the burst was stopped due to time limit, not completion - expect(result).toBe(false); // Should return false since provider never returned done - expect(duration).toBeGreaterThanOrEqual(100); // Should have run for at least the burst length - expect(duration).toBeLessThan(200); // But not too much longer (allowing for timing variance) + expect(result).toBe(false); + expect(duration).toBeGreaterThanOrEqual(100); + expect(duration).toBeLessThan(200); expect(mockProvider.next).toHaveBeenCalledTimes(callCount); - expect(callCount).toBeGreaterThan(1); // Should have made multiple calls before stopping - // Verify that burst was terminated due to time limit (not normal completion) - expect(mockLogger.info).toHaveBeenCalledWith( - expect.stringContaining('burst ending after'), - ); + expect(callCount).toBeGreaterThan(1); }); it('should complete burst normally when provider returns done before burst length', async () => { @@ -123,7 +119,7 @@ describe('IncrementalIngestionEngine - Burst Length', () => { provider: mockProvider, manager: mockManager, connection: mockConnection, - burstLength: { seconds: 10 }, // Long burst length + burstLength: { seconds: 10 }, restLength: { minutes: 1 }, logger: mockLogger, ready: Promise.resolve(), @@ -150,25 +146,66 @@ describe('IncrementalIngestionEngine - Burst Length', () => { }); const signal = new AbortController().signal; + const start = performance.now(); const result = await engine.ingestOneBurst('test-ingestion', signal); + const duration = performance.now() - start; - // Should return true when provider indicates done expect(result).toBe(true); expect(mockProvider.next).toHaveBeenCalledTimes(1); + expect(duration).toBeLessThan(100); // Should complete quickly since provider returns done immediately + }); - // Should NOT log the burst exceeded message - expect(mockLogger.info).not.toHaveBeenCalledWith( - expect.stringContaining('burst ending after'), - ); + it('should stop burst when time limit is reached', async () => { + const mockProvider = createMockProvider(); + const mockManager = createMockManager(); + const mockConnection = createMockConnection(); + const mockLogger = createMockLogger(); - // Should log burst initiation and completion - expect(mockLogger.info).toHaveBeenCalledWith( - "incremental-engine: Ingestion 'test-ingestion' burst initiated", - ); - expect(mockLogger.info).toHaveBeenCalledWith( - expect.stringContaining( - "incremental-engine: Ingestion 'test-ingestion' burst complete", - ), - ); + const options: IterationEngineOptions = { + provider: mockProvider, + manager: mockManager, + connection: mockConnection, + burstLength: { milliseconds: 80 }, + restLength: { minutes: 1 }, + logger: mockLogger, + ready: Promise.resolve(), + }; + + const engine = new IncrementalIngestionEngine(options); + + let callCount = 0; + mockProvider.around.mockImplementation(async fn => { + await fn({}); + }); + + mockProvider.next.mockImplementation(async () => { + callCount++; + await new Promise(resolve => setTimeout(resolve, 30)); + return { + done: false, + entities: [ + { + entity: { + kind: 'Component', + metadata: { name: `test-component-${callCount}` }, + }, + }, + ], + cursor: `cursor-${callCount}`, + }; + }); + + const signal = new AbortController().signal; + const start = performance.now(); + + const result = await engine.ingestOneBurst('test-ingestion', signal); + + const duration = performance.now() - start; + + expect(result).toBe(false); + expect(mockProvider.next).toHaveBeenCalledTimes(3); + expect(callCount).toBe(3); + expect(duration).toBeGreaterThanOrEqual(90); + expect(duration).toBeLessThan(120); }); });