paulononato/backstage
1
0
Fork 0
Code Issues Pull Requests Actions 22 Packages Projects Releases Wiki Activity

Merge pull request #6027 from backstage/rugvip/pipeline

Browse Source 
catalog-backend: introduce TaskPipeline
This commit is contained in:
Patrik Oldsberg
2021-06-15 11:24:49 +02:00
committed by GitHub
parent 0176d78f27 72fbf43721
commit 3be9904e63
4 changed files with 329 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/fuzzy-jobs-relate.md
+5
View File
@@ -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.
plugins/catalog-backend/src/next/DefaultCatalogProcessingEngine.ts
+82 -89
View File
@@ -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<RefreshStateItem>({
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<string>([
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<string>([
stringifyEntityRef(result.completedEntity),
...result.relations.map(relation =>
stringifyEntityRef(relation.source),
),
]);
await this.stitcher.stitch(setOfThingsToStitch);
}),
);
}
private async wait() {
await new Promise<void>(resolve => setTimeout(resolve, 1000));
},
});
}
async stop() {
this.running = false;
if (this.stopFunc) {
this.stopFunc();
this.stopFunc = undefined;
}
}
}
plugins/catalog-backend/src/next/TaskPipeline.test.ts
+121
View File
@@ -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<number>();
const processedTasks = new Array<number>();
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<number>({
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<number>({
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<number>({
loadTasks,
processTask,
lowWatermark: 2,
highWatermark: 4,
});
const stop2 = startTaskPipeline<number>({
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<number>({
loadTasks: async () => [],
processTask: async () => {},
lowWatermark: 3,
highWatermark: 3,
});
}).toThrow('must be lower');
});
});
plugins/catalog-backend/src/next/TaskPipeline.ts
+121
View File
@@ -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<T> = {
/**
* 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<Array<T>>;
/**
* 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<void>;
/**
* 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<T>(options: Options<T>) {
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);
};
}