diff --git a/.changeset/slimy-kids-jam.md b/.changeset/slimy-kids-jam.md new file mode 100644 index 0000000000..c02daf325e --- /dev/null +++ b/.changeset/slimy-kids-jam.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend': minor +--- + +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..bed1fe6f50 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, trace } from '@opentelemetry/api'; import { ProcessingDatabase, RefreshStateItem } from '../database/types'; import { createCounterMetric, createSummaryMetric } from '../util/metrics'; import { @@ -35,9 +35,16 @@ import { Stitcher } from '../stitching/Stitcher'; import { startTaskPipeline } from './TaskPipeline'; import { PluginTaskScheduler } from '@backstage/backend-tasks'; import { Config } from '@backstage/config'; +import { + addEntityAttributes, + TRACER_ID, + withActiveSpan, +} from '../util/opentelemetry'; const CACHE_TTL = 5; +const tracer = trace.getTracer(TRACER_ID); + export type ProgressTracker = ReturnType; export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { @@ -131,177 +138,181 @@ export class DefaultCatalogProcessingEngine implements CatalogProcessingEngine { } }, processTask: async item => { - const track = this.tracker.processStart(item, this.logger); + await withActiveSpan(tracer, 'ProcessingRun', async span => { + const track = this.tracker.processStart(item, this.logger); + addEntityAttributes(span, item.unprocessedEntity); - 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(); + 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(); + 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); } - - 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); - } + }); }, }); } 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, }; diff --git a/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts b/plugins/catalog-backend/src/processing/DefaultCatalogProcessingOrchestrator.ts index d177a7d4a8..c5eaacbc68 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, trace } from '@opentelemetry/api'; import { Entity, EntityPolicy, @@ -55,6 +56,13 @@ import { } from './util'; import { CatalogRulesEnforcer } from '../ingestion/CatalogRules'; import { ProcessorCacheManager } from './ProcessorCacheManager'; +import { + addEntityAttributes, + TRACER_ID, + withActiveSpan, +} from '../util/opentelemetry'; + +const tracer = trace.getTracer(TRACER_ID); type Context = { entityRef: string; @@ -64,6 +72,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 @@ -179,54 +199,71 @@ export class DefaultCatalogProcessingOrchestrator entity: Entity, context: Context, ): Promise { - let res = entity; + return await withActiveSpan(tracer, 'ProcessingStage', async stageSpan => { + addEntityAttributes(stageSpan, entity); + stageSpan.setAttribute('backstage.catalog.processor.stage', 'preProcess'); + let res = entity; - 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, - ); + for (const processor of this.options.processors) { + if (processor.preProcessEntity) { + let innerRes = res; + res = await withActiveSpan(tracer, 'ProcessingStep', async span => { + addEntityAttributes(span, entity); + 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; + }); } /** * 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, entity); + stageSpan.setAttribute( + 'backstage.catalog.processor.stage', + 'enforcePolicy', ); - } + 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; + }); } /** @@ -236,50 +273,62 @@ 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, 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 + 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, entity); + addProcessorAttributes(span, 'validateEntityKind', 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`, + ); + } + }); } /** @@ -289,65 +338,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, entity); + stageSpan.setAttribute( + 'backstage.catalog.processor.stage', + 'readLocation', ); + 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, entity); + addProcessorAttributes(span, 'readLocation', 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}`, - ); - } - } + }); } /** @@ -357,26 +422,39 @@ export class DefaultCatalogProcessingOrchestrator entity: Entity, context: Context, ): Promise { - let res = entity; + return await withActiveSpan(tracer, 'ProcessingStage', async stageSpan => { + addEntityAttributes(stageSpan, entity); + stageSpan.setAttribute( + 'backstage.catalog.processor.stage', + 'postProcessEntity', + ); + let res = entity; - 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, - ); + for (const processor of this.options.processors) { + if (processor.postProcessEntity) { + let innerRes = res; + res = await withActiveSpan(tracer, 'ProcessingStep', async span => { + addEntityAttributes(span, entity); + 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; + }); } } diff --git a/plugins/catalog-backend/src/util/opentelemetry.ts b/plugins/catalog-backend/src/util/opentelemetry.ts new file mode 100644 index 0000000000..3fb83d7909 --- /dev/null +++ b/plugins/catalog-backend/src/util/opentelemetry.ts @@ -0,0 +1,101 @@ +/* + * 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, SpanOptions, SpanStatusCode, Tracer } from '@opentelemetry/api'; +import { Entity } from '@backstage/catalog-model'; + +export const TRACER_ID = 'backstage-plugin-catalog-backend'; + +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 + +const onException = (e: Error, span: Span) => { + span.recordException(e); + span.setStatus({ + code: SpanStatusCode.ERROR, + }); +}; + +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); + + // if fn is an async function attach a recordException and spanEnd callback to the promise + if (isPromiseLike(ret)) { + ret.then( + () => { + span.end(); + }, + e => { + onException(e, span); + span.end(); + }, + ); + } else { + span.end(); + } + + return ret; + } 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); + }); +}