diff --git a/.changeset/late-eyes-serve.md b/.changeset/late-eyes-serve.md new file mode 100644 index 0000000000..37d8e7a22a --- /dev/null +++ b/.changeset/late-eyes-serve.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend': patch +--- + +Wrap single `pipelineLoop` of TaskPipeline in a span for better traces diff --git a/plugins/catalog-backend/src/processing/TaskPipeline.ts b/plugins/catalog-backend/src/processing/TaskPipeline.ts index 44030be9e0..98cf221b6e 100644 --- a/plugins/catalog-backend/src/processing/TaskPipeline.ts +++ b/plugins/catalog-backend/src/processing/TaskPipeline.ts @@ -14,7 +14,11 @@ * limitations under the License. */ +import { TRACER_ID, withActiveSpan } from '../util/opentelemetry'; +import { trace } from '@opentelemetry/api'; + const DEFAULT_POLLING_INTERVAL_MS = 1000; +const tracer = trace.getTracer(TRACER_ID); type Options = { /** @@ -85,34 +89,36 @@ export function startTaskPipeline(options: Options) { async function pipelineLoop() { while (!abortSignal.aborted) { - if (state.inFlightCount <= lowWatermark) { - const loadCount = highWatermark - state.inFlightCount; - const loadedItems = await Promise.resolve() - .then(() => loadTasks(loadCount)) - .catch(() => { - // Silently swallow errors and go back to sleep to try again; we - // delegate to the loadTasks function itself to catch errors and log - // if it so desires - return []; - }); - if (loadedItems.length && !abortSignal.aborted) { - state.inFlightCount += loadedItems.length; - for (const item of loadedItems) { - Promise.resolve() - .then(() => processTask(item)) - .catch(() => { - // Silently swallow errors and go back to sleep to try again; we - // delegate to the processTask function itself to catch errors - // and log if it so desires - }) - .finally(() => { - state.inFlightCount -= 1; - barrier.release(); - }); + await withActiveSpan(tracer, 'TaskPipelineLoop', async span => { + if (state.inFlightCount <= lowWatermark) { + const loadCount = highWatermark - state.inFlightCount; + const loadedItems = await Promise.resolve(loadCount) + .then(loadTasks) + .catch(() => { + // Silently swallow errors and go back to sleep to try again; we + // delegate to the loadTasks function itself to catch errors and log + // if it so desires + return []; + }); + span.setAttribute('itemCount', loadedItems.length); + if (loadedItems.length && !abortSignal.aborted) { + state.inFlightCount += loadedItems.length; + for (const item of loadedItems) { + Promise.resolve(item) + .then(processTask) + .catch(() => { + // Silently swallow errors and go back to sleep to try again; we + // delegate to the processTask function itself to catch errors + // and log if it so desires + }) + .finally(() => { + state.inFlightCount -= 1; + barrier.release(); + }); + } } } - } - + }); await barrier.wait(); } }