From f32252cdf631378a398d4afbd0d785c53535fe3a Mon Sep 17 00:00:00 2001 From: Mike Bryant Date: Wed, 26 Apr 2023 19:55:05 +0100 Subject: [PATCH 01/12] feat(catalog-backend): Add observability for catalog processing Signed-off-by: Mike Bryant Signed-off-by: Mike Bryant --- .changeset/slimy-kids-jam.md | 5 + .../DefaultCatalogProcessingEngine.ts | 320 +++++++++--------- .../DefaultCatalogProcessingOrchestrator.ts | 90 +++-- .../catalog-backend/src/util/opentelemetry.ts | 32 ++ 4 files changed, 266 insertions(+), 181 deletions(-) create mode 100644 .changeset/slimy-kids-jam.md create mode 100644 plugins/catalog-backend/src/util/opentelemetry.ts diff --git a/.changeset/slimy-kids-jam.md b/.changeset/slimy-kids-jam.md new file mode 100644 index 0000000000..ea2897bed1 --- /dev/null +++ b/.changeset/slimy-kids-jam.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend': patch +--- + +Added OpenTelemetry spans for catalog processing diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts index 8e2f8b2f6f..e9f87c36c7 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts @@ -23,7 +23,7 @@ import { assertError, serializeError, stringifyError } from '@backstage/errors'; import { Hash } from 'crypto'; import stableStringify from 'fast-json-stable-stringify'; import { Logger } from 'winston'; -import { metrics } from '@opentelemetry/api'; +import { metrics, SpanStatusCode, trace } from '@opentelemetry/api'; import { ProcessingDatabase, RefreshStateItem } from '../database/types'; import { createCounterMetric, createSummaryMetric } from '../util/metrics'; import { @@ -35,9 +35,12 @@ import { Stitcher } from '../stitching/Stitcher'; import { startTaskPipeline } from './TaskPipeline'; import { PluginTaskScheduler } from '@backstage/backend-tasks'; import { Config } from '@backstage/config'; +import { addEntityAttributes, TRACER_ID } from '../util/opentelemetry'; const CACHE_TTL = 5; +const tracer = trace.getTracer(TRACER_ID); + export type ProgressTracker = ReturnType; export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { @@ -131,177 +134,186 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { } }, processTask: async item => { - const track = this.tracker.processStart(item, this.logger); + await tracer.startActiveSpan('ProcessingRun', async span => { + const track = this.tracker.processStart(item, this.logger); + addEntityAttributes(span, item.entityRef); - try { - const { - id, - state, - unprocessedEntity, - entityRef, - locationKey, - resultHash: previousResultHash, - } = item; - const result = await this.orchestrator.process({ - entity: unprocessedEntity, - state, - }); + try { + const { + id, + state, + unprocessedEntity, + entityRef, + locationKey, + resultHash: previousResultHash, + } = item; + const result = await this.orchestrator.process({ + entity: unprocessedEntity, + state, + }); - track.markProcessorsCompleted(result); + track.markProcessorsCompleted(result); - if (result.ok) { - const { ttl: _, ...stateWithoutTtl } = state ?? {}; - if ( - stableStringify(stateWithoutTtl) !== stableStringify(result.state) - ) { + if (result.ok) { + const { ttl: _, ...stateWithoutTtl } = state ?? {}; + if ( + stableStringify(stateWithoutTtl) !== stableStringify(result.state) + ) { + await this.processingDatabase.transaction(async tx => { + await this.processingDatabase.updateEntityCache(tx, { + id, + state: { + ttl: CACHE_TTL, + ...result.state, + }, + }); + }); + } + } else { + const maybeTtl = state?.ttl; + const ttl = Number.isInteger(maybeTtl) ? (maybeTtl as number) : 0; await this.processingDatabase.transaction(async tx => { await this.processingDatabase.updateEntityCache(tx, { id, - state: { - ttl: CACHE_TTL, - ...result.state, - }, + state: ttl > 0 ? { ...state, ttl: ttl - 1 } : {}, }); }); } - } else { - const maybeTtl = state?.ttl; - const ttl = Number.isInteger(maybeTtl) ? (maybeTtl as number) : 0; - await this.processingDatabase.transaction(async tx => { - await this.processingDatabase.updateEntityCache(tx, { - id, - state: ttl > 0 ? { ...state, ttl: ttl - 1 } : {}, + + const location = + unprocessedEntity?.metadata?.annotations?.[ANNOTATION_LOCATION]; + for (const error of result.errors) { + this.logger.warn(error.message, { + entity: entityRef, + location, }); - }); - } + } + const errorsString = JSON.stringify( + result.errors.map(e => serializeError(e)), + ); - const location = - unprocessedEntity?.metadata?.annotations?.[ANNOTATION_LOCATION]; - for (const error of result.errors) { - this.logger.warn(error.message, { - entity: entityRef, - location, - }); - } - const errorsString = JSON.stringify( - result.errors.map(e => serializeError(e)), - ); + let hashBuilder = this.createHash().update(errorsString); - let hashBuilder = this.createHash().update(errorsString); - - if (result.ok) { - const { entityRefs: parents } = - await this.processingDatabase.transaction(tx => - this.processingDatabase.listParents(tx, { - entityRef, - }), - ); - - hashBuilder = hashBuilder - .update(stableStringify({ ...result.completedEntity })) - .update(stableStringify([...result.deferredEntities])) - .update(stableStringify([...result.relations])) - .update(stableStringify([...result.refreshKeys])) - .update(stableStringify([...parents])); - } - - const resultHash = hashBuilder.digest('hex'); - if (resultHash === previousResultHash) { - // If nothing changed in our produced outputs, we cannot have any - // significant effect on our surroundings; therefore, we just abort - // without any updates / stitching. - track.markSuccessfulWithNoChanges(); - return; - } - - // 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) { - // notify the error listener if the entity can not be processed. - Promise.resolve(undefined) - .then(() => - this.onProcessingError?.({ - unprocessedEntity, - errors: result.errors, - }), - ) - .catch(error => { - this.logger.debug( - `Processing error listener threw an exception, ${stringifyError( - error, - )}`, + if (result.ok) { + const { entityRefs: parents } = + await this.processingDatabase.transaction(tx => + this.processingDatabase.listParents(tx, { + entityRef, + }), ); - }); + hashBuilder = hashBuilder + .update(stableStringify({ ...result.completedEntity })) + .update(stableStringify([...result.deferredEntities])) + .update(stableStringify([...result.relations])) + .update(stableStringify([...result.refreshKeys])) + .update(stableStringify([...parents])); + } + + const resultHash = hashBuilder.digest('hex'); + if (resultHash === previousResultHash) { + // If nothing changed in our produced outputs, we cannot have any + // significant effect on our surroundings; therefore, we just abort + // without any updates / stitching. + track.markSuccessfulWithNoChanges(); + span.end(); + return; + } + + // 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) { + // notify the error listener if the entity can not be processed. + Promise.resolve(undefined) + .then(() => + this.onProcessingError?.({ + unprocessedEntity, + errors: result.errors, + }), + ) + .catch(error => { + this.logger.debug( + `Processing error listener threw an exception, ${stringifyError( + error, + )}`, + ); + }); + + await this.processingDatabase.transaction(async tx => { + await this.processingDatabase.updateProcessedEntityErrors(tx, { + id, + errors: errorsString, + resultHash, + }); + }); + await this.stitcher.stitch( + new Set([stringifyEntityRef(unprocessedEntity)]), + ); + track.markSuccessfulWithErrors(); + span.setStatus({ code: SpanStatusCode.ERROR }); + span.end(); + return; + } + + result.completedEntity.metadata.uid = id; + let oldRelationSources: Map; await this.processingDatabase.transaction(async tx => { - await this.processingDatabase.updateProcessedEntityErrors(tx, { - id, - errors: errorsString, - resultHash, - }); + const { previous } = + await this.processingDatabase.updateProcessedEntity(tx, { + id, + processedEntity: result.completedEntity, + resultHash, + errors: errorsString, + relations: result.relations, + deferredEntities: result.deferredEntities, + locationKey, + refreshKeys: result.refreshKeys, + }); + oldRelationSources = new Map( + previous.relations.map(r => [ + `${r.source_entity_ref}:${r.type}`, + r.source_entity_ref, + ]), + ); }); - await this.stitcher.stitch( - new Set([stringifyEntityRef(unprocessedEntity)]), + + const newRelationSources = new Map( + result.relations.map(relation => { + const sourceEntityRef = stringifyEntityRef(relation.source); + return [`${sourceEntityRef}:${relation.type}`, sourceEntityRef]; + }), ); - track.markSuccessfulWithErrors(); - return; + + const setOfThingsToStitch = new Set([ + stringifyEntityRef(result.completedEntity), + ]); + newRelationSources.forEach((sourceEntityRef, uniqueKey) => { + if (!oldRelationSources.has(uniqueKey)) { + setOfThingsToStitch.add(sourceEntityRef); + } + }); + oldRelationSources!.forEach((sourceEntityRef, uniqueKey) => { + if (!newRelationSources.has(uniqueKey)) { + setOfThingsToStitch.add(sourceEntityRef); + } + }); + + await this.stitcher.stitch(setOfThingsToStitch); + + track.markSuccessfulWithChanges(setOfThingsToStitch.size); + } catch (error) { + assertError(error); + track.markFailed(error); + span.recordException(error); + span.setStatus({ code: SpanStatusCode.ERROR }); } - - result.completedEntity.metadata.uid = id; - let oldRelationSources: Map; - await this.processingDatabase.transaction(async tx => { - const { previous } = - await this.processingDatabase.updateProcessedEntity(tx, { - id, - processedEntity: result.completedEntity, - resultHash, - errors: errorsString, - relations: result.relations, - deferredEntities: result.deferredEntities, - locationKey, - refreshKeys: result.refreshKeys, - }); - oldRelationSources = new Map( - previous.relations.map(r => [ - `${r.source_entity_ref}:${r.type}`, - r.source_entity_ref, - ]), - ); - }); - - const newRelationSources = new Map( - result.relations.map(relation => { - const sourceEntityRef = stringifyEntityRef(relation.source); - return [`${sourceEntityRef}:${relation.type}`, sourceEntityRef]; - }), - ); - - const setOfThingsToStitch = new Set([ - stringifyEntityRef(result.completedEntity), - ]); - newRelationSources.forEach((sourceEntityRef, uniqueKey) => { - if (!oldRelationSources.has(uniqueKey)) { - setOfThingsToStitch.add(sourceEntityRef); - } - }); - oldRelationSources!.forEach((sourceEntityRef, uniqueKey) => { - if (!newRelationSources.has(uniqueKey)) { - setOfThingsToStitch.add(sourceEntityRef); - } - }); - - await this.stitcher.stitch(setOfThingsToStitch); - - track.markSuccessfulWithChanges(setOfThingsToStitch.size); - } catch (error) { - assertError(error); - track.markFailed(error); - } + span.end(); + }); }, }); } diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts index d177a7d4a8..2b4102677e 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts @@ -14,6 +14,7 @@ * limitations under the License. */ +import { Span, SpanStatusCode, trace } from '@opentelemetry/api'; import { Entity, EntityPolicy, @@ -55,6 +56,9 @@ import { } from './util'; import { CatalogRulesEnforcer } from '../ingestion/CatalogRules'; import { ProcessorCacheManager } from './ProcessorCacheManager'; +import { addEntityAttributes, TRACER_ID } from '../util/opentelemetry'; + +const tracer = trace.getTracer(TRACER_ID); type Context = { entityRef: string; @@ -64,6 +68,18 @@ type Context = { cache: ProcessorCacheManager; }; +function addProcessorAttributes( + span: Span, + stage: string, + processor: CatalogProcessor, +) { + span.setAttribute('backstage.catalog.processor.stage', stage); + span.setAttribute( + 'backstage.catalog.processor.name', + processor.getProcessorName(), + ); +} + /** @public */ export class DefaultCatalogProcessingOrchestrator implements CatalogProcessingOrchestrator @@ -183,20 +199,30 @@ export class DefaultCatalogProcessingOrchestrator for (const processor of this.options.processors) { if (processor.preProcessEntity) { - try { - res = await processor.preProcessEntity( - res, - context.location, - context.collector.forProcessor(processor), - context.originLocation, - context.cache.forProcessor(processor), - ); - } catch (e) { - throw new InputError( - `Processor ${processor.constructor.name} threw an error while preprocessing`, - e, - ); - } + let innerRes = res; + res = await tracer.startActiveSpan('ProcessingStep', async span => { + addEntityAttributes(span, context.entityRef); + addProcessorAttributes(span, 'preProcessEntity', processor); + try { + innerRes = await processor.preProcessEntity!( + innerRes, + context.location, + context.collector.forProcessor(processor), + context.originLocation, + context.cache.forProcessor(processor), + ); + } catch (e) { + span.recordException(e); + span.setStatus({ code: SpanStatusCode.ERROR }); + span.end(); + throw new InputError( + `Processor ${processor.constructor.name} threw an error while preprocessing`, + e, + ); + } + span.end(); + return innerRes; + }); } } @@ -361,19 +387,29 @@ export class DefaultCatalogProcessingOrchestrator for (const processor of this.options.processors) { if (processor.postProcessEntity) { - try { - res = await processor.postProcessEntity( - res, - context.location, - context.collector.forProcessor(processor), - context.cache.forProcessor(processor), - ); - } catch (e) { - throw new InputError( - `Processor ${processor.constructor.name} threw an error while postprocessing`, - e, - ); - } + let innerRes = res; + res = await tracer.startActiveSpan('ProcessingStep', async span => { + addEntityAttributes(span, context.entityRef); + addProcessorAttributes(span, 'postProcessEntity', processor); + try { + innerRes = await processor.postProcessEntity!( + innerRes, + context.location, + context.collector.forProcessor(processor), + context.cache.forProcessor(processor), + ); + } catch (e) { + span.recordException(e); + span.setStatus({ code: SpanStatusCode.ERROR }); + span.end(); + throw new InputError( + `Processor ${processor.constructor.name} threw an error while postprocessing`, + e, + ); + } + span.end(); + return innerRes; + }); } } diff --git a/plugins/catalog-backend/src/util/opentelemetry.ts b/plugins/catalog-backend/src/util/opentelemetry.ts new file mode 100644 index 0000000000..16ec276a79 --- /dev/null +++ b/plugins/catalog-backend/src/util/opentelemetry.ts @@ -0,0 +1,32 @@ +/* + * Copyright 2023 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 { Span, SpanStatusCode } from '@opentelemetry/api'; +import { parseEntityRef } from '@backstage/catalog-model'; + +export const TRACER_ID = 'backstage-plugin-catalog-backend'; + +export function addEntityAttributes(span: Span, entityRef: string) { + try { + const fields = parseEntityRef(entityRef); + span.setAttribute('backstage.entity.kind', fields.kind); + span.setAttribute('backstage.entity.namespace', fields.namespace); + span.setAttribute('backstage.entity.name', fields.name); + } catch (err) { + span.recordException(err); + span.setStatus({ code: SpanStatusCode.ERROR }); + } +} From c1d8f44180bf5a30eaff492d8b1f8ecf97d31241 Mon Sep 17 00:00:00 2001 From: Mike Bryant Date: Thu, 18 May 2023 21:01:35 +0100 Subject: [PATCH 02/12] refactor: Pull out withActiveSpan to ensure we always end the span regardless of exception Signed-off-by: Mike Bryant --- .../DefaultCatalogProcessingEngine.ts | 19 ++++---- .../DefaultCatalogProcessingOrchestrator.ts | 18 +++---- .../catalog-backend/src/util/opentelemetry.ts | 47 ++++++++++++++++++- 3 files changed, 62 insertions(+), 22 deletions(-) diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts index e9f87c36c7..2ae104666f 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts @@ -23,7 +23,7 @@ import { assertError, serializeError, stringifyError } from '@backstage/errors'; import { Hash } from 'crypto'; import stableStringify from 'fast-json-stable-stringify'; import { Logger } from 'winston'; -import { metrics, SpanStatusCode, trace } from '@opentelemetry/api'; +import { metrics, trace } from '@opentelemetry/api'; import { ProcessingDatabase, RefreshStateItem } from '../database/types'; import { createCounterMetric, createSummaryMetric } from '../util/metrics'; import { @@ -35,7 +35,11 @@ import { Stitcher } from '../stitching/Stitcher'; import { startTaskPipeline } from './TaskPipeline'; import { PluginTaskScheduler } from '@backstage/backend-tasks'; import { Config } from '@backstage/config'; -import { addEntityAttributes, TRACER_ID } from '../util/opentelemetry'; +import { + addEntityAttributes, + TRACER_ID, + withActiveSpan, +} from '../util/opentelemetry'; const CACHE_TTL = 5; @@ -134,7 +138,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { } }, processTask: async item => { - await tracer.startActiveSpan('ProcessingRun', async span => { + await withActiveSpan(tracer, 'ProcessingRun', async span => { const track = this.tracker.processStart(item, this.logger); addEntityAttributes(span, item.entityRef); @@ -157,7 +161,8 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { if (result.ok) { const { ttl: _, ...stateWithoutTtl } = state ?? {}; if ( - stableStringify(stateWithoutTtl) !== stableStringify(result.state) + stableStringify(stateWithoutTtl) !== + stableStringify(result.state) ) { await this.processingDatabase.transaction(async tx => { await this.processingDatabase.updateEntityCache(tx, { @@ -216,7 +221,6 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { // significant effect on our surroundings; therefore, we just abort // without any updates / stitching. track.markSuccessfulWithNoChanges(); - span.end(); return; } @@ -255,8 +259,6 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { new Set([stringifyEntityRef(unprocessedEntity)]), ); track.markSuccessfulWithErrors(); - span.setStatus({ code: SpanStatusCode.ERROR }); - span.end(); return; } @@ -309,10 +311,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { } catch (error) { assertError(error); track.markFailed(error); - span.recordException(error); - span.setStatus({ code: SpanStatusCode.ERROR }); } - span.end(); }); }, }); diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts index 2b4102677e..8091c6f120 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts @@ -56,7 +56,11 @@ import { } from './util'; import { CatalogRulesEnforcer } from '../ingestion/CatalogRules'; import { ProcessorCacheManager } from './ProcessorCacheManager'; -import { addEntityAttributes, TRACER_ID } from '../util/opentelemetry'; +import { + addEntityAttributes, + TRACER_ID, + withActiveSpan, +} from '../util/opentelemetry'; const tracer = trace.getTracer(TRACER_ID); @@ -200,7 +204,7 @@ export class DefaultCatalogProcessingOrchestrator for (const processor of this.options.processors) { if (processor.preProcessEntity) { let innerRes = res; - res = await tracer.startActiveSpan('ProcessingStep', async span => { + res = await withActiveSpan(tracer, 'ProcessingStep', async span => { addEntityAttributes(span, context.entityRef); addProcessorAttributes(span, 'preProcessEntity', processor); try { @@ -212,15 +216,11 @@ export class DefaultCatalogProcessingOrchestrator context.cache.forProcessor(processor), ); } catch (e) { - span.recordException(e); - span.setStatus({ code: SpanStatusCode.ERROR }); - span.end(); throw new InputError( `Processor ${processor.constructor.name} threw an error while preprocessing`, e, ); } - span.end(); return innerRes; }); } @@ -388,7 +388,7 @@ export class DefaultCatalogProcessingOrchestrator for (const processor of this.options.processors) { if (processor.postProcessEntity) { let innerRes = res; - res = await tracer.startActiveSpan('ProcessingStep', async span => { + res = await withActiveSpan(tracer, 'ProcessingStep', async span => { addEntityAttributes(span, context.entityRef); addProcessorAttributes(span, 'postProcessEntity', processor); try { @@ -399,15 +399,11 @@ export class DefaultCatalogProcessingOrchestrator context.cache.forProcessor(processor), ); } catch (e) { - span.recordException(e); - span.setStatus({ code: SpanStatusCode.ERROR }); - span.end(); throw new InputError( `Processor ${processor.constructor.name} threw an error while postprocessing`, e, ); } - span.end(); return innerRes; }); } diff --git a/plugins/catalog-backend/src/util/opentelemetry.ts b/plugins/catalog-backend/src/util/opentelemetry.ts index 16ec276a79..391869deb8 100644 --- a/plugins/catalog-backend/src/util/opentelemetry.ts +++ b/plugins/catalog-backend/src/util/opentelemetry.ts @@ -14,7 +14,7 @@ * limitations under the License. */ -import { Span, SpanStatusCode } from '@opentelemetry/api'; +import { Span, SpanOptions, SpanStatusCode, Tracer } from '@opentelemetry/api'; import { parseEntityRef } from '@backstage/catalog-model'; export const TRACER_ID = 'backstage-plugin-catalog-backend'; @@ -30,3 +30,48 @@ export function addEntityAttributes(span: Span, entityRef: string) { span.setStatus({ code: SpanStatusCode.ERROR }); } } + +// Adapted from https://github.com/open-telemetry/opentelemetry-js/blob/359fbcc40a859057a02b14e84599eac399b8dba7/api/src/trace/SugaredTracer.ts +// While waiting for something like https://github.com/open-telemetry/opentelemetry-js/pull/3317 to land upstream + +const onException = (e: Error, span: Span) => { + span.recordException(e); + span.setStatus({ + code: SpanStatusCode.ERROR, + }); +}; + +function handleFn ReturnType>( + span: Span, + fn: F, +): ReturnType { + try { + const ret = fn(span) as Promise>; + // if fn is an async function attach a recordException and spanEnd callback to the promise + if (typeof ret.then === 'function' && typeof ret.catch === 'function') { + return ret + .catch((e: Error) => { + onException(e, span); + throw e; + }) + .finally(() => span.end()) as ReturnType; + } + span.end(); + return ret as ReturnType; + } catch (e) { + onException(e, span); + span.end(); + throw e; + } +} + +export function withActiveSpan ReturnType>( + tracer: Tracer, + name: string, + fn: F, + spanOptions: SpanOptions = {}, +): ReturnType { + return tracer.startActiveSpan(name, spanOptions, (span: Span) => { + return handleFn(span, fn); + }); +} From db2cac67450cd75c95ce1ed761b2c6f0993f7bde Mon Sep 17 00:00:00 2001 From: Mike Bryant Date: Thu, 18 May 2023 21:06:51 +0100 Subject: [PATCH 03/12] feat: Add ProcessingStage span Signed-off-by: Mike Bryant --- .../DefaultCatalogProcessingOrchestrator.ts | 112 ++++++++++-------- 1 file changed, 63 insertions(+), 49 deletions(-) diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts index 8091c6f120..8b4f4127f3 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts @@ -199,34 +199,41 @@ export class DefaultCatalogProcessingOrchestrator entity: Entity, context: Context, ): Promise { - let res = entity; + return await withActiveSpan(tracer, 'ProcessingStage', async stageSpan => { + addEntityAttributes(stageSpan, context.entityRef); + stageSpan.setAttribute( + 'backstage.catalog.processor.stage', + 'preProcessEntity', + ); + let res = entity; - for (const processor of this.options.processors) { - if (processor.preProcessEntity) { - let innerRes = res; - res = await withActiveSpan(tracer, 'ProcessingStep', async span => { - addEntityAttributes(span, context.entityRef); - addProcessorAttributes(span, 'preProcessEntity', processor); - try { - innerRes = await processor.preProcessEntity!( - innerRes, - context.location, - context.collector.forProcessor(processor), - context.originLocation, - context.cache.forProcessor(processor), - ); - } catch (e) { - throw new InputError( - `Processor ${processor.constructor.name} threw an error while preprocessing`, - e, - ); - } - return innerRes; - }); + for (const processor of this.options.processors) { + if (processor.preProcessEntity) { + let innerRes = res; + res = await withActiveSpan(tracer, 'ProcessingStep', async span => { + addEntityAttributes(span, context.entityRef); + addProcessorAttributes(span, 'preProcessEntity', processor); + try { + innerRes = await processor.preProcessEntity!( + innerRes, + context.location, + context.collector.forProcessor(processor), + context.originLocation, + context.cache.forProcessor(processor), + ); + } catch (e) { + throw new InputError( + `Processor ${processor.constructor.name} threw an error while preprocessing`, + e, + ); + } + return innerRes; + }); + } } - } - return res; + return res; + }); } /** @@ -383,32 +390,39 @@ export class DefaultCatalogProcessingOrchestrator entity: Entity, context: Context, ): Promise { - let res = entity; + return await withActiveSpan(tracer, 'ProcessingStage', async stageSpan => { + addEntityAttributes(stageSpan, context.entityRef); + stageSpan.setAttribute( + 'backstage.catalog.processor.stage', + 'postProcessEntity', + ); + let res = entity; - for (const processor of this.options.processors) { - if (processor.postProcessEntity) { - let innerRes = res; - res = await withActiveSpan(tracer, 'ProcessingStep', async span => { - addEntityAttributes(span, context.entityRef); - addProcessorAttributes(span, 'postProcessEntity', processor); - try { - innerRes = await processor.postProcessEntity!( - innerRes, - context.location, - context.collector.forProcessor(processor), - context.cache.forProcessor(processor), - ); - } catch (e) { - throw new InputError( - `Processor ${processor.constructor.name} threw an error while postprocessing`, - e, - ); - } - return innerRes; - }); + for (const processor of this.options.processors) { + if (processor.postProcessEntity) { + let innerRes = res; + res = await withActiveSpan(tracer, 'ProcessingStep', async span => { + addEntityAttributes(span, context.entityRef); + addProcessorAttributes(span, 'postProcessEntity', processor); + try { + innerRes = await processor.postProcessEntity!( + innerRes, + context.location, + context.collector.forProcessor(processor), + context.cache.forProcessor(processor), + ); + } catch (e) { + throw new InputError( + `Processor ${processor.constructor.name} threw an error while postprocessing`, + e, + ); + } + return innerRes; + }); + } } - } - return res; + return res; + }); } } From dae8956754cad162bf20f49c50a62b65be87e4ac Mon Sep 17 00:00:00 2001 From: Mike Bryant Date: Thu, 18 May 2023 21:15:41 +0100 Subject: [PATCH 04/12] feat: Add tracing to validate stage Signed-off-by: Mike Bryant --- .../DefaultCatalogProcessingOrchestrator.ts | 85 +++++++++++-------- 1 file changed, 50 insertions(+), 35 deletions(-) diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts index 8b4f4127f3..aabe703806 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts @@ -269,50 +269,65 @@ export class DefaultCatalogProcessingOrchestrator entity: Entity, context: Context, ): Promise { - // Double check that none of the previous steps tried to change something - // related to the entity ref, which would break downstream - if (stringifyEntityRef(entity) !== context.entityRef) { - throw new ConflictError( - 'Fatal: The entity kind, namespace, or name changed during processing', + return await withActiveSpan(tracer, 'ProcessingStage', async stageSpan => { + addEntityAttributes(stageSpan, context.entityRef); + stageSpan.setAttribute( + 'backstage.catalog.processor.stage', + 'validateEntity', ); - } + // Double check that none of the previous steps tried to change something + // related to the entity ref, which would break downstream + if (stringifyEntityRef(entity) !== context.entityRef) { + throw new ConflictError( + 'Fatal: The entity kind, namespace, or name changed during processing', + ); + } - // Validate that the end result is a valid Entity at all - try { - validateEntity(entity); - } catch (e) { - throw new ConflictError( - `Entity envelope for ${context.entityRef} failed validation after preprocessing`, - e, - ); - } + // Validate that the end result is a valid Entity at all + try { + validateEntity(entity); + } catch (e) { + throw new ConflictError( + `Entity envelope for ${context.entityRef} failed validation after preprocessing`, + e, + ); + } - let valid = false; + let valid = false; - for (const processor of this.options.processors) { - if (processor.validateEntityKind) { - try { - const thisValid = await processor.validateEntityKind(entity); - if (thisValid) { - valid = true; - if (this.options.legacySingleProcessorValidation) { - break; + for (const processor of this.options.processors) { + if (processor.validateEntityKind) { + try { + const thisValid = await withActiveSpan( + tracer, + 'ProcessingStep', + async span => { + addEntityAttributes(span, context.entityRef); + addProcessorAttributes(span, 'postProcessEntity', processor); + return await processor.validateEntityKind(entity); + }, + ); + if (thisValid) { + valid = true; + if (this.options.legacySingleProcessorValidation) { + break; + } } + } catch (e) { + throw new InputError( + `Processor ${processor.constructor.name} threw an error while validating the entity ${context.entityRef}`, + e, + ); } - } catch (e) { - throw new InputError( - `Processor ${processor.constructor.name} threw an error while validating the entity ${context.entityRef}`, - e, - ); } } - } - if (!valid) { - throw new InputError( - `No processor recognized the entity ${context.entityRef} as valid, possibly caused by a foreign kind or apiVersion`, - ); - } + if (!valid) { + throw new InputError( + `No processor recognized the entity ${context.entityRef} as valid, possibly caused by a foreign kind or apiVersion`, + ); + } + }); } /** From 9550ae9ad4378b850f3c4c922244eace5b1c3d2c Mon Sep 17 00:00:00 2001 From: Mike Bryant Date: Thu, 18 May 2023 21:19:11 +0100 Subject: [PATCH 05/12] feat: Add span for policy stage Signed-off-by: Mike Bryant --- .../DefaultCatalogProcessingOrchestrator.ts | 41 +++++++++++-------- 1 file changed, 24 insertions(+), 17 deletions(-) diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts index aabe703806..6fa44db1a6 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts @@ -240,26 +240,33 @@ export class DefaultCatalogProcessingOrchestrator * Enforce entity policies making sure that entities conform to a general schema */ private async runPolicyStep(entity: Entity): Promise { - let policyEnforcedEntity: Entity | undefined; - - try { - policyEnforcedEntity = await this.options.policy.enforce(entity); - } catch (e) { - throw new InputError( - `Policy check failed for ${stringifyEntityRef(entity)}`, - e, + return await withActiveSpan(tracer, 'ProcessingStage', async stageSpan => { + addEntityAttributes(stageSpan, stringifyEntityRef(entity)); + stageSpan.setAttribute( + 'backstage.catalog.processor.stage', + 'enforcePolicyEntity', ); - } + let policyEnforcedEntity: Entity | undefined; - if (!policyEnforcedEntity) { - throw new Error( - `Policy unexpectedly returned no data for ${stringifyEntityRef( - entity, - )}`, - ); - } + try { + policyEnforcedEntity = await this.options.policy.enforce(entity); + } catch (e) { + throw new InputError( + `Policy check failed for ${stringifyEntityRef(entity)}`, + e, + ); + } - return policyEnforcedEntity; + if (!policyEnforcedEntity) { + throw new Error( + `Policy unexpectedly returned no data for ${stringifyEntityRef( + entity, + )}`, + ); + } + + return policyEnforcedEntity; + }); } /** From bb5ab706d7e6d8413bf74d7243e5de70a5331e47 Mon Sep 17 00:00:00 2001 From: Mike Bryant Date: Thu, 18 May 2023 21:24:50 +0100 Subject: [PATCH 06/12] feat: Add spans for read steps Signed-off-by: Mike Bryant --- .../DefaultCatalogProcessingOrchestrator.ts | 122 ++++++++++-------- 1 file changed, 69 insertions(+), 53 deletions(-) diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts index 6fa44db1a6..91a1c2f9a6 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts @@ -344,65 +344,81 @@ export class DefaultCatalogProcessingOrchestrator entity: LocationEntity, context: Context, ): Promise { - const { type = context.location.type, presence = 'required' } = entity.spec; - const targets = new Array(); - if (entity.spec.target) { - targets.push(entity.spec.target); - } - if (entity.spec.targets) { - targets.push(...entity.spec.targets); - } - - for (const maybeRelativeTarget of targets) { - if (type === 'file' && maybeRelativeTarget.endsWith(path.sep)) { - context.collector.generic()( - processingResult.inputError( - context.location, - `LocationEntityProcessor cannot handle ${type} type location with target ${context.location.target} that ends with a path separator`, - ), - ); - continue; - } - const target = toAbsoluteUrl( - this.options.integrations, - context.location, - type, - maybeRelativeTarget, + return await withActiveSpan(tracer, 'ProcessingStage', async stageSpan => { + addEntityAttributes(stageSpan, context.entityRef); + stageSpan.setAttribute( + 'backstage.catalog.processor.stage', + 'readLocationEntity', ); + const { type = context.location.type, presence = 'required' } = + entity.spec; + const targets = new Array(); + if (entity.spec.target) { + targets.push(entity.spec.target); + } + if (entity.spec.targets) { + targets.push(...entity.spec.targets); + } - let didRead = false; - for (const processor of this.options.processors) { - if (processor.readLocation) { - try { - const read = await processor.readLocation( - { - type, - target, - presence, - }, - presence === 'optional', - context.collector.forProcessor(processor), - this.options.parser, - context.cache.forProcessor(processor, target), - ); - if (read) { - didRead = true; - break; + for (const maybeRelativeTarget of targets) { + if (type === 'file' && maybeRelativeTarget.endsWith(path.sep)) { + context.collector.generic()( + processingResult.inputError( + context.location, + `LocationEntityProcessor cannot handle ${type} type location with target ${context.location.target} that ends with a path separator`, + ), + ); + continue; + } + const target = toAbsoluteUrl( + this.options.integrations, + context.location, + type, + maybeRelativeTarget, + ); + + let didRead = false; + for (const processor of this.options.processors) { + if (processor.readLocation) { + try { + const read = await withActiveSpan( + tracer, + 'ProcessingStep', + async span => { + addEntityAttributes(span, context.entityRef); + addProcessorAttributes(span, 'readLocationEntity', processor); + return await processor.readLocation( + { + type, + target, + presence, + }, + presence === 'optional', + context.collector.forProcessor(processor), + this.options.parser, + context.cache.forProcessor(processor, target), + ); + }, + ); + if (read) { + didRead = true; + break; + } + } catch (e) { + throw new InputError( + `Processor ${processor.constructor.name} threw an error while reading ${type}:${target}`, + e, + ); } - } catch (e) { - throw new InputError( - `Processor ${processor.constructor.name} threw an error while reading ${type}:${target}`, - e, - ); } } + if (!didRead) { + throw new InputError( + `No processor was able to handle reading of ${type}:${target}`, + ); + } } - if (!didRead) { - throw new InputError( - `No processor was able to handle reading of ${type}:${target}`, - ); - } - } + }); } /** From 5060c6ed7d04b9875a4c50b37a5b33d467e45a69 Mon Sep 17 00:00:00 2001 From: Mike Bryant Date: Thu, 18 May 2023 22:52:00 +0100 Subject: [PATCH 07/12] fix: Fix tsc errors Signed-off-by: Mike Bryant --- .../src/processing/DefaultCatalogProcessingOrchestrator.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts index 91a1c2f9a6..dfca5d2058 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts @@ -14,7 +14,7 @@ * limitations under the License. */ -import { Span, SpanStatusCode, trace } from '@opentelemetry/api'; +import { Span, trace } from '@opentelemetry/api'; import { Entity, EntityPolicy, @@ -311,7 +311,7 @@ export class DefaultCatalogProcessingOrchestrator async span => { addEntityAttributes(span, context.entityRef); addProcessorAttributes(span, 'postProcessEntity', processor); - return await processor.validateEntityKind(entity); + return await processor.validateEntityKind!(entity); }, ); if (thisValid) { @@ -387,7 +387,7 @@ export class DefaultCatalogProcessingOrchestrator async span => { addEntityAttributes(span, context.entityRef); addProcessorAttributes(span, 'readLocationEntity', processor); - return await processor.readLocation( + return await processor.readLocation!( { type, target, From ee48005f1871de089ec9eedb85faf0caf3290b5f Mon Sep 17 00:00:00 2001 From: Mike Bryant Date: Thu, 18 May 2023 23:15:24 +0100 Subject: [PATCH 08/12] fix: Supply objects that satisfy the type constraints in tests Signed-off-by: Mike Bryant --- .../processing/DefaultCatalogProcessingOrchestrator.test.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.test.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.test.ts index ceb6992d7a..fc6855f6df 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.test.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.test.ts @@ -194,10 +194,12 @@ describe('DefaultCatalogProcessingOrchestrator', () => { it('runs all processor validations when asked to', async () => { const validate = jest.fn(async () => true); - const processor1: Partial = { + const processor1: CatalogProcessor = { + getProcessorName: () => 'processor1', validateEntityKind: validate, }; - const processor2: Partial = { + const processor2: CatalogProcessor = { + getProcessorName: () => 'processor2', validateEntityKind: validate, }; From f40c8ac5347c0bd145eb1f7c0dc978d9c17f0148 Mon Sep 17 00:00:00 2001 From: Mike Bryant Date: Tue, 13 Jun 2023 23:33:50 +0100 Subject: [PATCH 09/12] style: Apply review comments Signed-off-by: Mike Bryant --- .../DefaultCatalogProcessingOrchestrator.ts | 18 ++++++------------ 1 file changed, 6 insertions(+), 12 deletions(-) diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts index dfca5d2058..98f9cd32a5 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts @@ -201,10 +201,7 @@ export class DefaultCatalogProcessingOrchestrator ): Promise { return await withActiveSpan(tracer, 'ProcessingStage', async stageSpan => { addEntityAttributes(stageSpan, context.entityRef); - stageSpan.setAttribute( - 'backstage.catalog.processor.stage', - 'preProcessEntity', - ); + stageSpan.setAttribute('backstage.catalog.processor.stage', 'preProcess'); let res = entity; for (const processor of this.options.processors) { @@ -244,7 +241,7 @@ export class DefaultCatalogProcessingOrchestrator addEntityAttributes(stageSpan, stringifyEntityRef(entity)); stageSpan.setAttribute( 'backstage.catalog.processor.stage', - 'enforcePolicyEntity', + 'enforcePolicy', ); let policyEnforcedEntity: Entity | undefined; @@ -278,10 +275,7 @@ export class DefaultCatalogProcessingOrchestrator ): Promise { return await withActiveSpan(tracer, 'ProcessingStage', async stageSpan => { addEntityAttributes(stageSpan, context.entityRef); - stageSpan.setAttribute( - 'backstage.catalog.processor.stage', - 'validateEntity', - ); + stageSpan.setAttribute('backstage.catalog.processor.stage', 'validate'); // Double check that none of the previous steps tried to change something // related to the entity ref, which would break downstream if (stringifyEntityRef(entity) !== context.entityRef) { @@ -310,7 +304,7 @@ export class DefaultCatalogProcessingOrchestrator 'ProcessingStep', async span => { addEntityAttributes(span, context.entityRef); - addProcessorAttributes(span, 'postProcessEntity', processor); + addProcessorAttributes(span, 'validateEntityKind', processor); return await processor.validateEntityKind!(entity); }, ); @@ -348,7 +342,7 @@ export class DefaultCatalogProcessingOrchestrator addEntityAttributes(stageSpan, context.entityRef); stageSpan.setAttribute( 'backstage.catalog.processor.stage', - 'readLocationEntity', + 'readLocation', ); const { type = context.location.type, presence = 'required' } = entity.spec; @@ -386,7 +380,7 @@ export class DefaultCatalogProcessingOrchestrator 'ProcessingStep', async span => { addEntityAttributes(span, context.entityRef); - addProcessorAttributes(span, 'readLocationEntity', processor); + addProcessorAttributes(span, 'readLocation', processor); return await processor.readLocation!( { type, From ab51b0c9aee7ee298b99f32aedba573f251e49ee Mon Sep 17 00:00:00 2001 From: Mike Bryant Date: Wed, 14 Jun 2023 00:09:02 +0100 Subject: [PATCH 10/12] style: Apply review comments Signed-off-by: Mike Bryant --- .changeset/slimy-kids-jam.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/slimy-kids-jam.md b/.changeset/slimy-kids-jam.md index ea2897bed1..c02daf325e 100644 --- a/.changeset/slimy-kids-jam.md +++ b/.changeset/slimy-kids-jam.md @@ -1,5 +1,5 @@ --- -'@backstage/plugin-catalog-backend': patch +'@backstage/plugin-catalog-backend': minor --- Added OpenTelemetry spans for catalog processing From 73a5205447dccd188558549145dc180562e2c52d Mon Sep 17 00:00:00 2001 From: Mike Bryant Date: Wed, 5 Jul 2023 18:47:52 +0100 Subject: [PATCH 11/12] fix: Use real entity instead of ref for attributes Signed-off-by: Mike Bryant --- .../DefaultCatalogProcessingEngine.ts | 2 +- .../DefaultCatalogProcessingOrchestrator.ts | 18 ++++++------ .../catalog-backend/src/util/opentelemetry.ts | 29 ++++++++++++------- 3 files changed, 29 insertions(+), 20 deletions(-) diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts index 2ae104666f..bed1fe6f50 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingEngine.ts @@ -140,7 +140,7 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { processTask: async item => { await withActiveSpan(tracer, 'ProcessingRun', async span => { const track = this.tracker.processStart(item, this.logger); - addEntityAttributes(span, item.entityRef); + addEntityAttributes(span, item.unprocessedEntity); try { const { diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts index 98f9cd32a5..c5eaacbc68 100644 --- a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts +++ b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts @@ -200,7 +200,7 @@ export class DefaultCatalogProcessingOrchestrator context: Context, ): Promise { return await withActiveSpan(tracer, 'ProcessingStage', async stageSpan => { - addEntityAttributes(stageSpan, context.entityRef); + addEntityAttributes(stageSpan, entity); stageSpan.setAttribute('backstage.catalog.processor.stage', 'preProcess'); let res = entity; @@ -208,7 +208,7 @@ export class DefaultCatalogProcessingOrchestrator if (processor.preProcessEntity) { let innerRes = res; res = await withActiveSpan(tracer, 'ProcessingStep', async span => { - addEntityAttributes(span, context.entityRef); + addEntityAttributes(span, entity); addProcessorAttributes(span, 'preProcessEntity', processor); try { innerRes = await processor.preProcessEntity!( @@ -238,7 +238,7 @@ export class DefaultCatalogProcessingOrchestrator */ private async runPolicyStep(entity: Entity): Promise { return await withActiveSpan(tracer, 'ProcessingStage', async stageSpan => { - addEntityAttributes(stageSpan, stringifyEntityRef(entity)); + addEntityAttributes(stageSpan, entity); stageSpan.setAttribute( 'backstage.catalog.processor.stage', 'enforcePolicy', @@ -274,7 +274,7 @@ export class DefaultCatalogProcessingOrchestrator context: Context, ): Promise { return await withActiveSpan(tracer, 'ProcessingStage', async stageSpan => { - addEntityAttributes(stageSpan, context.entityRef); + addEntityAttributes(stageSpan, entity); stageSpan.setAttribute('backstage.catalog.processor.stage', 'validate'); // Double check that none of the previous steps tried to change something // related to the entity ref, which would break downstream @@ -303,7 +303,7 @@ export class DefaultCatalogProcessingOrchestrator tracer, 'ProcessingStep', async span => { - addEntityAttributes(span, context.entityRef); + addEntityAttributes(span, entity); addProcessorAttributes(span, 'validateEntityKind', processor); return await processor.validateEntityKind!(entity); }, @@ -339,7 +339,7 @@ export class DefaultCatalogProcessingOrchestrator context: Context, ): Promise { return await withActiveSpan(tracer, 'ProcessingStage', async stageSpan => { - addEntityAttributes(stageSpan, context.entityRef); + addEntityAttributes(stageSpan, entity); stageSpan.setAttribute( 'backstage.catalog.processor.stage', 'readLocation', @@ -379,7 +379,7 @@ export class DefaultCatalogProcessingOrchestrator tracer, 'ProcessingStep', async span => { - addEntityAttributes(span, context.entityRef); + addEntityAttributes(span, entity); addProcessorAttributes(span, 'readLocation', processor); return await processor.readLocation!( { @@ -423,7 +423,7 @@ export class DefaultCatalogProcessingOrchestrator context: Context, ): Promise { return await withActiveSpan(tracer, 'ProcessingStage', async stageSpan => { - addEntityAttributes(stageSpan, context.entityRef); + addEntityAttributes(stageSpan, entity); stageSpan.setAttribute( 'backstage.catalog.processor.stage', 'postProcessEntity', @@ -434,7 +434,7 @@ export class DefaultCatalogProcessingOrchestrator if (processor.postProcessEntity) { let innerRes = res; res = await withActiveSpan(tracer, 'ProcessingStep', async span => { - addEntityAttributes(span, context.entityRef); + addEntityAttributes(span, entity); addProcessorAttributes(span, 'postProcessEntity', processor); try { innerRes = await processor.postProcessEntity!( diff --git a/plugins/catalog-backend/src/util/opentelemetry.ts b/plugins/catalog-backend/src/util/opentelemetry.ts index 391869deb8..c2df008452 100644 --- a/plugins/catalog-backend/src/util/opentelemetry.ts +++ b/plugins/catalog-backend/src/util/opentelemetry.ts @@ -15,22 +15,31 @@ */ import { Span, SpanOptions, SpanStatusCode, Tracer } from '@opentelemetry/api'; -import { parseEntityRef } from '@backstage/catalog-model'; +import { Entity } from '@backstage/catalog-model'; export const TRACER_ID = 'backstage-plugin-catalog-backend'; -export function addEntityAttributes(span: Span, entityRef: string) { - try { - const fields = parseEntityRef(entityRef); - span.setAttribute('backstage.entity.kind', fields.kind); - span.setAttribute('backstage.entity.namespace', fields.namespace); - span.setAttribute('backstage.entity.name', fields.name); - } catch (err) { - span.recordException(err); - span.setStatus({ code: SpanStatusCode.ERROR }); +function setAttributeIfDefined(span: Span, attribute: string, value?: string) { + if (value !== null && value !== undefined) { + span.setAttribute(attribute, value); } } +export function addEntityAttributes(span: Span, entity: Entity) { + setAttributeIfDefined(span, 'backstage.entity.apiVersion', entity.apiVersion); + setAttributeIfDefined(span, 'backstage.entity.kind', entity.kind); + setAttributeIfDefined( + span, + 'backstage.entity.metadata.namespace', + entity.metadata?.namespace, + ); + setAttributeIfDefined( + span, + 'backstage.entity.metadata.name', + entity.metadata?.name, + ); +} + // Adapted from https://github.com/open-telemetry/opentelemetry-js/blob/359fbcc40a859057a02b14e84599eac399b8dba7/api/src/trace/SugaredTracer.ts // While waiting for something like https://github.com/open-telemetry/opentelemetry-js/pull/3317 to land upstream From b9379b45bb1b296377cb624b22eda32713c3a2df Mon Sep 17 00:00:00 2001 From: Mike Bryant Date: Fri, 21 Jul 2023 19:11:40 +0100 Subject: [PATCH 12/12] chore: Use PromiseLike instead of Promise Follow review comments, use PromiseLike with a type guard Signed-off-by: Mike Bryant --- .../catalog-backend/src/util/opentelemetry.ts | 33 ++++++++++++++----- 1 file changed, 24 insertions(+), 9 deletions(-) diff --git a/plugins/catalog-backend/src/util/opentelemetry.ts b/plugins/catalog-backend/src/util/opentelemetry.ts index c2df008452..3fb83d7909 100644 --- a/plugins/catalog-backend/src/util/opentelemetry.ts +++ b/plugins/catalog-backend/src/util/opentelemetry.ts @@ -50,23 +50,38 @@ const onException = (e: Error, span: Span) => { }); }; +function isPromiseLike(obj: PromiseLike | S): obj is PromiseLike { + return ( + !!obj && + (typeof obj === 'object' || typeof obj === 'function') && + 'then' in obj && + typeof obj.then === 'function' + ); +} + function handleFn ReturnType>( span: Span, fn: F, ): ReturnType { try { - const ret = fn(span) as Promise>; + const ret = fn(span); + // if fn is an async function attach a recordException and spanEnd callback to the promise - if (typeof ret.then === 'function' && typeof ret.catch === 'function') { - return ret - .catch((e: Error) => { + if (isPromiseLike(ret)) { + ret.then( + () => { + span.end(); + }, + e => { onException(e, span); - throw e; - }) - .finally(() => span.end()) as ReturnType; + span.end(); + }, + ); + } else { + span.end(); } - span.end(); - return ret as ReturnType; + + return ret; } catch (e) { onException(e, span); span.end();