diff --git a/.changeset/fuzzy-jobs-relate.md b/.changeset/fuzzy-jobs-relate.md new file mode 100644 index 0000000000..6828d6c3c0 --- /dev/null +++ b/.changeset/fuzzy-jobs-relate.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend': patch +--- + +Switches the default catalog processing engine to use a batched streaming task execution strategy for higher parallelism. diff --git a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts index 5de012e936..7b8cabcd84 100644 --- a/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts @@ -21,9 +21,10 @@ import { } from '@backstage/catalog-model'; import { serializeError } from '@backstage/errors'; import { Logger } from 'winston'; -import { ProcessingDatabase } from './database/types'; +import { ProcessingDatabase, RefreshStateItem } from './database/types'; import { CatalogProcessingOrchestrator } from './processing/types'; import { Stitcher } from './stitching/Stitcher'; +import { startTaskPipeline } from './TaskPipeline'; import { CatalogProcessingEngine, EntityProvider, @@ -80,7 +81,7 @@ class Connection implements EntityProviderConnection { } export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { - private running = false; + private stopFunc?: () => void; constructor( private readonly logger: Logger, @@ -99,106 +100,98 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { }), ); } - this.running = true; - this.run(); - } - private async run() { - while (this.running) { - try { - // TODO: We want to disconnect the queue popping and message processing - // so that if the queue popping fails we exponentially back off in order to give the DB room to sort itself out. - await this.process(); - } catch (e) { - this.logger.warn('Processing failed with:', e); - // TODO: this can be a little smarter as mentioned in the above comment. - // But for now, if something fails, wait a brief time to pick up the next message. - await this.wait(); - } - } - } - - private async process() { - const { items } = await this.processingDatabase.transaction(async tx => { - return this.processingDatabase.getProcessableEntities(tx, { - processBatchSize: 1, - }); - }); - - if (!items.length) { - // No items to process, wait and try again. - await this.wait(); - return; + if (this.stopFunc) { + throw new Error('Processing engine is already started'); } - // TODO: replace Promise.all with something more sophisticated for parallel processing. - await Promise.all( - items.map(async item => { - const { id, state, unprocessedEntity, entityRef } = item; - const result = await this.orchestrator.process({ - entity: unprocessedEntity, - state, - }); - - for (const error of result.errors) { - // TODO(freben): Try to extract the location out of the unprocessed - // entity and add as meta to the log lines - this.logger.warn(error.message, { - entity: entityRef, - }); + this.stopFunc = startTaskPipeline({ + lowWatermark: 5, + highWatermark: 10, + loadTasks: async count => { + try { + const { items } = await this.processingDatabase.transaction( + async tx => { + return this.processingDatabase.getProcessableEntities(tx, { + processBatchSize: count, + }); + }, + ); + return items; + } catch (error) { + this.logger.warn('Failed to load processing items', error); + return []; } - const errorsString = JSON.stringify( - result.errors.map(e => serializeError(e)), - ); + }, + processTask: async item => { + try { + const { id, state, unprocessedEntity, entityRef } = item; + const result = await this.orchestrator.process({ + entity: unprocessedEntity, + state, + }); - // If the result was marked as not OK, it signals that some part of the - // processing pipeline threw an exception. This can happen both as part of - // non-catastrophic things such as due to validation errors, as well as if - // something fatal happens inside the processing for other reasons. In any - // case, this means we can't trust that anything in the output is okay. So - // just store the errors and trigger a stich so that they become visible to - // the outside. - if (!result.ok) { + for (const error of result.errors) { + // TODO(freben): Try to extract the location out of the unprocessed + // entity and add as meta to the log lines + this.logger.warn(error.message, { + entity: entityRef, + }); + } + const errorsString = JSON.stringify( + result.errors.map(e => serializeError(e)), + ); + + // If the result was marked as not OK, it signals that some part of the + // processing pipeline threw an exception. This can happen both as part of + // non-catastrophic things such as due to validation errors, as well as if + // something fatal happens inside the processing for other reasons. In any + // case, this means we can't trust that anything in the output is okay. So + // just store the errors and trigger a stich so that they become visible to + // the outside. + if (!result.ok) { + await this.processingDatabase.transaction(async tx => { + await this.processingDatabase.updateProcessedEntityErrors(tx, { + id, + errors: errorsString, + }); + }); + await this.stitcher.stitch( + new Set([stringifyEntityRef(unprocessedEntity)]), + ); + return; + } + + result.completedEntity.metadata.uid = id; await this.processingDatabase.transaction(async tx => { - await this.processingDatabase.updateProcessedEntityErrors(tx, { + await this.processingDatabase.updateProcessedEntity(tx, { id, + processedEntity: result.completedEntity, + state: result.state, errors: errorsString, + relations: result.relations, + deferredEntities: result.deferredEntities, }); }); - await this.stitcher.stitch( - new Set([stringifyEntityRef(unprocessedEntity)]), - ); - return; + + const setOfThingsToStitch = new Set([ + stringifyEntityRef(result.completedEntity), + ...result.relations.map(relation => + stringifyEntityRef(relation.source), + ), + ]); + await this.stitcher.stitch(setOfThingsToStitch); + } catch (error) { + this.logger.warn('Processing failed with:', error); } - - result.completedEntity.metadata.uid = id; - await this.processingDatabase.transaction(async tx => { - await this.processingDatabase.updateProcessedEntity(tx, { - id, - processedEntity: result.completedEntity, - state: result.state, - errors: errorsString, - relations: result.relations, - deferredEntities: result.deferredEntities, - }); - }); - - const setOfThingsToStitch = new Set([ - stringifyEntityRef(result.completedEntity), - ...result.relations.map(relation => - stringifyEntityRef(relation.source), - ), - ]); - await this.stitcher.stitch(setOfThingsToStitch); - }), - ); - } - - private async wait() { - await new Promise(resolve => setTimeout(resolve, 1000)); + }, + }); } async stop() { - this.running = false; + if (this.stopFunc) { + this.stopFunc(); + this.stopFunc = undefined; + } } } diff --git a/plugins/catalog-backend/src/next/TaskPipeline.test.ts b/plugins/catalog-backend/src/next/TaskPipeline.test.ts new file mode 100644 index 0000000000..ddf668767a --- /dev/null +++ b/plugins/catalog-backend/src/next/TaskPipeline.test.ts @@ -0,0 +1,121 @@ +/* + * Copyright 2021 Spotify AB + * + * 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 { startTaskPipeline } from './TaskPipeline'; + +function createLimitedLoader(count: number, loadDelay?: number) { + const items = new Array(count).fill(0).map((_, index) => index); + const loadCounts = new Array(); + const processedTasks = new Array(); + + let resolveDone: (_: { + loadCounts: number[]; + processedTasks: number[]; + }) => void; + const done = new Promise<{ loadCounts: number[]; processedTasks: number[] }>( + resolve => { + resolveDone = resolve; + }, + ); + + const loadTasks = async (loadCount: number) => { + if (loadCounts.length < (loadDelay || 0)) { + loadCounts.push(0); + return []; + } + const loadedItems = items.splice(0, loadCount); + loadCounts.push(loadedItems.length); + return loadedItems; + }; + const processTask = async (item: number) => { + processedTasks.push(item); + await new Promise(resolve => setTimeout(resolve)); // emulate a bit of work + if (processedTasks.length === count) { + resolveDone({ processedTasks, loadCounts }); + } + }; + + return { loadTasks, processTask, done }; +} + +describe('startTaskPipeline', () => { + it('should process some tasks', async () => { + const { loadTasks, processTask, done } = createLimitedLoader(6); + const stop = startTaskPipeline({ + loadTasks, + processTask, + lowWatermark: 1, + highWatermark: 3, + }); + + const { loadCounts, processedTasks } = await done; + stop(); + + expect(loadCounts).toEqual([3, 2, 1]); + expect(processedTasks).toEqual([0, 1, 2, 3, 4, 5]); + }); + + it('should pick up processing after it runs dry', async () => { + const { loadTasks, processTask, done } = createLimitedLoader(5, 2); + const stop = startTaskPipeline({ + loadTasks, + processTask, + lowWatermark: 2, + highWatermark: 3, + pollingIntervalMs: 1, + }); + + const { loadCounts, processedTasks } = await done; + stop(); + + expect(loadCounts).toEqual([0, 0, 3, 1, 1]); + expect(processedTasks).toEqual([0, 1, 2, 3, 4]); + }); + + it('should process in parallel', async () => { + const { loadTasks, processTask, done } = createLimitedLoader(13); + const stop1 = startTaskPipeline({ + loadTasks, + processTask, + lowWatermark: 2, + highWatermark: 4, + }); + const stop2 = startTaskPipeline({ + loadTasks, + processTask, + lowWatermark: 2, + highWatermark: 4, + }); + + const { loadCounts, processedTasks } = await done; + stop1(); + stop2(); + + expect(loadCounts).toEqual([4, 4, 2, 2, 1]); + expect(processedTasks).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]); + }); + + it('should require lowWatermark to be lower than highWatermark', async () => { + expect(() => { + startTaskPipeline({ + loadTasks: async () => [], + processTask: async () => {}, + lowWatermark: 3, + highWatermark: 3, + }); + }).toThrow('must be lower'); + }); +}); diff --git a/plugins/catalog-backend/src/next/TaskPipeline.ts b/plugins/catalog-backend/src/next/TaskPipeline.ts new file mode 100644 index 0000000000..7021330626 --- /dev/null +++ b/plugins/catalog-backend/src/next/TaskPipeline.ts @@ -0,0 +1,121 @@ +/* + * Copyright 2021 Spotify AB + * + * 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. + */ + +const DEFAULT_POLLING_INTERVAL_MS = 1000; + +type Options = { + /** + * The callback used to load in new tasks. The number of items returned + * in the array must be at most `count` number of items, but may be lower. + * + * Any error thrown from this method fill be treated as an unhandled rejection. + */ + loadTasks: (count: number) => Promise>; + + /** + * The callback used to process a single item. + * + * Any error thrown from this method fill be treated as an unhandled rejection. + */ + processTask: (item: T) => Promise; + + /** + * The target minimum number of items to process in parallel. Once the number + * of in-flight tasks reaches this count, more tasks will be loaded in. + */ + lowWatermark: number; + + /** + * The maximum number of items to process in parallel. + */ + highWatermark: number; + + /** + * The interval at which tasks are polled for in the background when + * there aren't enough tasks to load to satisfy the low watermark. + * + * @default 1000 + */ + pollingIntervalMs?: number; +}; + +/** + * Creates a task processing pipeline which continuously loads in tasks to + * keep the number of parallel in-flight tasks between a low and high watermark. + * + * @param options The options for the pipeline. + * @returns A stop function which when called halts all processing. + */ +export function startTaskPipeline(options: Options) { + const { + loadTasks, + processTask, + lowWatermark, + highWatermark, + pollingIntervalMs = DEFAULT_POLLING_INTERVAL_MS, + } = options; + + if (lowWatermark >= highWatermark) { + throw new Error('lowWatermark must be lower than highWatermark'); + } + + let loading = false; + let stopped = false; + let inFlightCount = 0; + + async function maybeLoadMore() { + if (stopped || loading || inFlightCount > lowWatermark) { + return; + } + + // Once we hit the low watermark we load in enough items to reach the high watermark + loading = true; + const loadCount = highWatermark - inFlightCount; + const loadedItems = await loadTasks(loadCount); + loading = false; + + // We might not reach the high watermark here, in case there weren't enough items to load + inFlightCount += loadedItems.length; + loadedItems.forEach(item => { + processTask(item).finally(() => { + if (stopped) { + return; + } + + // For each item we complete we check if it's time to load more + inFlightCount -= 1; + maybeLoadMore(); + }); + }); + + // We might have processed some tasks while we where loading, so check if we can load more + if (loadedItems.length > 1) { + maybeLoadMore(); + } + } + + // This interval makes sure that we load in new items if the loop runs + // dry because of the lack of available tasks. As long as there are + // enough items to process this will be a noop. + const intervalId = setInterval(() => { + maybeLoadMore(); + }, pollingIntervalMs); + + return () => { + stopped = true; + clearInterval(intervalId); + }; +}