From 3447c0edf16f4d99d5dfe2b9e7e5226fb9d18478 Mon Sep 17 00:00:00 2001 From: Siim Kallas Date: Wed, 17 Jul 2024 18:17:07 +0300 Subject: [PATCH 01/14] sqs: use links instead of process spans --- .../package.json | 2 +- .../src/services/sqs.ts | 66 ++-- .../src/types.ts | 6 - .../test/sqs.test.ts | 291 ------------------ 4 files changed, 26 insertions(+), 339 deletions(-) diff --git a/plugins/node/opentelemetry-instrumentation-aws-sdk/package.json b/plugins/node/opentelemetry-instrumentation-aws-sdk/package.json index 30925579de..30c9211053 100644 --- a/plugins/node/opentelemetry-instrumentation-aws-sdk/package.json +++ b/plugins/node/opentelemetry-instrumentation-aws-sdk/package.json @@ -36,7 +36,7 @@ "prewatch": "npm run precompile", "prepublishOnly": "npm run compile", "tdd": "npm run test -- --watch-extensions ts --watch", - "test": "nyc ts-mocha -p tsconfig.json --require '@opentelemetry/contrib-test-utils' 'test/**/*.test.ts'", + "test": "nyc ts-mocha -p tsconfig.json --require '@opentelemetry/contrib-test-utils' 'test/**/sqs.test.ts'", "test-all-versions": "tav", "version:update": "node ../../../scripts/version-update.js", "watch": "tsc -w" diff --git a/plugins/node/opentelemetry-instrumentation-aws-sdk/src/services/sqs.ts b/plugins/node/opentelemetry-instrumentation-aws-sdk/src/services/sqs.ts index 07cedaa25d..64fc9e6e37 100644 --- a/plugins/node/opentelemetry-instrumentation-aws-sdk/src/services/sqs.ts +++ b/plugins/node/opentelemetry-instrumentation-aws-sdk/src/services/sqs.ts @@ -19,11 +19,9 @@ import { Span, propagation, trace, - context, ROOT_CONTEXT, Attributes, } from '@opentelemetry/api'; -import { pubsubPropagation } from '@opentelemetry/propagation-utils'; import { RequestMetadata, ServiceExtension } from './ServiceExtension'; import type { SQS } from 'aws-sdk'; import { @@ -33,7 +31,6 @@ import { } from '../types'; import { MESSAGINGDESTINATIONKINDVALUES_QUEUE, - MESSAGINGOPERATIONVALUES_PROCESS, MESSAGINGOPERATIONVALUES_RECEIVE, SEMATTRS_MESSAGING_DESTINATION, SEMATTRS_MESSAGING_DESTINATION_KIND, @@ -134,7 +131,7 @@ export class SqsServiceExtension implements ServiceExtension { responseHook = ( response: NormalizedResponse, span: Span, - tracer: Tracer, + _tracer: Tracer, config: AwsSdkInstrumentationConfig ) => { switch (response.request.commandName) { @@ -150,45 +147,32 @@ export class SqsServiceExtension implements ServiceExtension { break; case 'ReceiveMessage': { - const messages: SQS.Message[] = response?.data?.Messages; - if (messages) { - const queueUrl = this.extractQueueUrl(response.request.commandInput); - const queueName = this.extractQueueNameFromUrl(queueUrl); - - pubsubPropagation.patchMessagesArrayToStartProcessSpans({ - messages, - parentContext: trace.setSpan(context.active(), span), - tracer, - messageToSpanDetails: (message: SQS.Message) => ({ - name: queueName ?? 'unknown', - parentContext: propagation.extract( - ROOT_CONTEXT, - extractPropagationContext( - message, - config.sqsExtractContextPropagationFromPayload - ), - contextGetter - ), + console.log('ReceiveMessage'); + console.log(span); + const messages: SQS.Message[] = response?.data?.Messages || []; + + span.setAttribute('messaging.batch.message_count', messages.length); + + for (const message of messages) { + const propagatedContext = propagation.extract( + ROOT_CONTEXT, + extractPropagationContext( + message, + config.sqsExtractContextPropagationFromPayload + ), + contextGetter + ); + + const spanContext = trace.getSpanContext(propagatedContext); + + if (spanContext) { + span.addLink({ + context: spanContext, attributes: { - [SEMATTRS_MESSAGING_SYSTEM]: 'aws.sqs', - [SEMATTRS_MESSAGING_DESTINATION]: queueName, - [SEMATTRS_MESSAGING_DESTINATION_KIND]: - MESSAGINGDESTINATIONKINDVALUES_QUEUE, [SEMATTRS_MESSAGING_MESSAGE_ID]: message.MessageId, - [SEMATTRS_MESSAGING_URL]: queueUrl, - [SEMATTRS_MESSAGING_OPERATION]: - MESSAGINGOPERATIONVALUES_PROCESS, - }, - }), - processHook: (span: Span, message: SQS.Message) => - config.sqsProcessHook?.(span, { message }), - }); - - pubsubPropagation.patchArrayForProcessSpans( - messages, - tracer, - context.active() - ); + } + }) + } } break; } diff --git a/plugins/node/opentelemetry-instrumentation-aws-sdk/src/types.ts b/plugins/node/opentelemetry-instrumentation-aws-sdk/src/types.ts index ce99e8c441..8416b6211d 100644 --- a/plugins/node/opentelemetry-instrumentation-aws-sdk/src/types.ts +++ b/plugins/node/opentelemetry-instrumentation-aws-sdk/src/types.ts @@ -60,9 +60,6 @@ export interface AwsSdkResponseCustomAttributeFunction { export interface AwsSdkSqsProcessHookInformation { message: SQS.Message; } -export interface AwsSdkSqsProcessCustomAttributeFunction { - (span: Span, sqsProcessInfo: AwsSdkSqsProcessHookInformation): void; -} export type AwsSdkDynamoDBStatementSerializer = ( operation: string, @@ -76,9 +73,6 @@ export interface AwsSdkInstrumentationConfig extends InstrumentationConfig { /** hook for adding custom attributes when response is received from aws */ responseHook?: AwsSdkResponseCustomAttributeFunction; - /** hook for adding custom attribute when an sqs process span is started */ - sqsProcessHook?: AwsSdkSqsProcessCustomAttributeFunction; - /** custom serializer function for the db.statement attribute in DynamoDB spans */ dynamoDBStatementSerializer?: AwsSdkDynamoDBStatementSerializer; diff --git a/plugins/node/opentelemetry-instrumentation-aws-sdk/test/sqs.test.ts b/plugins/node/opentelemetry-instrumentation-aws-sdk/test/sqs.test.ts index e97d82bb2c..066357e134 100644 --- a/plugins/node/opentelemetry-instrumentation-aws-sdk/test/sqs.test.ts +++ b/plugins/node/opentelemetry-instrumentation-aws-sdk/test/sqs.test.ts @@ -27,7 +27,6 @@ import type { SQS } from 'aws-sdk'; import { MESSAGINGDESTINATIONKINDVALUES_QUEUE, - MESSAGINGOPERATIONVALUES_PROCESS, MESSAGINGOPERATIONVALUES_RECEIVE, SEMATTRS_HTTP_STATUS_CODE, SEMATTRS_MESSAGING_DESTINATION, @@ -172,208 +171,6 @@ describe('SQS', () => { }); }); - describe('process spans', () => { - let receivedMessages: Message[]; - - const createProcessChildSpan = (msgContext: any) => { - const processChildSpan = trace - .getTracerProvider() - .getTracer('default') - .startSpan(`child span of sqs processing span of msg ${msgContext}`); - processChildSpan.end(); - }; - - const expectReceiver2ProcessWithNChildrenEach = ( - spans: ReadableSpan[], - numChildPerProcessSpan: number - ) => { - const awsReceiveSpan = spans.filter( - s => - s.attributes[SEMATTRS_MESSAGING_OPERATION] === - MESSAGINGOPERATIONVALUES_RECEIVE - ); - expect(awsReceiveSpan.length).toBe(1); - - const processSpans = spans.filter( - s => - s.attributes[SEMATTRS_MESSAGING_OPERATION] === - MESSAGINGOPERATIONVALUES_PROCESS - ); - expect(processSpans.length).toBe(2); - expect(processSpans[0].parentSpanId).toStrictEqual( - awsReceiveSpan[0].spanContext().spanId - ); - expect(processSpans[1].parentSpanId).toStrictEqual( - awsReceiveSpan[0].spanContext().spanId - ); - - const processChildSpans = spans.filter(s => s.kind === SpanKind.INTERNAL); - expect(processChildSpans.length).toBe(2 * numChildPerProcessSpan); - for (let i = 0; i < numChildPerProcessSpan; i++) { - expect(processChildSpans[2 * i + 0].parentSpanId).toStrictEqual( - processSpans[0].spanContext().spanId - ); - expect(processChildSpans[2 * i + 1].parentSpanId).toStrictEqual( - processSpans[1].spanContext().spanId - ); - } - }; - - const expectReceiver2ProcessWith1ChildEach = (spans: ReadableSpan[]) => { - expectReceiver2ProcessWithNChildrenEach(spans, 1); - }; - - const expectReceiver2ProcessWith2ChildEach = (spans: ReadableSpan[]) => { - expectReceiver2ProcessWithNChildrenEach(spans, 2); - }; - - const contextKeyFromTest = Symbol('context key from test'); - const contextValueFromTest = 'context value from test'; - - beforeEach(async () => { - const sqs = new AWS.SQS(); - await context.with( - context.active().setValue(contextKeyFromTest, contextValueFromTest), - async () => { - const res = await sqs - .receiveMessage({ - QueueUrl: 'queue/url/for/unittests', - }) - .promise(); - receivedMessages = res.Messages!; - } - ); - }); - - it('should create processing child with forEach', async () => { - receivedMessages.forEach(msg => { - createProcessChildSpan(msg.Body); - }); - expectReceiver2ProcessWith1ChildEach(getTestSpans()); - }); - - it('should create processing child with map', async () => { - receivedMessages.map(msg => { - createProcessChildSpan(msg.Body); - }); - expectReceiver2ProcessWith1ChildEach(getTestSpans()); - }); - - it('should not fail when mapping to non-object type', async () => { - receivedMessages - .map(msg => 'map result is string') - .map(s => 'some other string'); - }); - - it('should not fail when mapping to undefined type', async () => { - receivedMessages.map(msg => undefined).map(s => 'some other string'); - }); - - it('should create one processing child when throws in map', async () => { - try { - receivedMessages.map(msg => { - createProcessChildSpan(msg.Body); - throw Error('error from array.map'); - }); - } catch (err) {} - - const processChildSpans = getTestSpans().filter( - s => s.kind === SpanKind.INTERNAL - ); - expect(processChildSpans.length).toBe(1); - }); - - it('should create processing child with two forEach', async () => { - receivedMessages.forEach(msg => { - createProcessChildSpan(msg.Body); - }); - receivedMessages.forEach(msg => { - createProcessChildSpan(msg.Body); - }); - expectReceiver2ProcessWith2ChildEach(getTestSpans()); - }); - - it('should forward all parameters to forEach callback', async () => { - const objectForThis = {}; - receivedMessages.forEach(function (this: any, msg, index, array) { - expect(msg).not.toBeUndefined(); - expect(index).toBeLessThan(2); - expect(index).toBeGreaterThanOrEqual(0); - expect(array).toBe(receivedMessages); - expect(this).toBe(objectForThis); - }, objectForThis); - }); - - it('should create one processing child with forEach that throws', async () => { - try { - receivedMessages.forEach(msg => { - createProcessChildSpan(msg.Body); - throw Error('error from forEach'); - }); - } catch (err) {} - const processChildSpans = getTestSpans().filter( - s => s.kind === SpanKind.INTERNAL - ); - expect(processChildSpans.length).toBe(1); - }); - - it.skip('should create processing child with array index access', async () => { - for (let i = 0; i < receivedMessages.length; i++) { - const msg = receivedMessages[i]; - createProcessChildSpan(msg.Body); - } - expectReceiver2ProcessWith1ChildEach(getTestSpans()); - }); - - it('should create processing child with map and forEach calls', async () => { - receivedMessages - .map(msg => ({ payload: msg.Body })) - .forEach(msgBody => { - createProcessChildSpan(msgBody); - }); - expectReceiver2ProcessWith1ChildEach(getTestSpans()); - }); - - it('should create processing child with filter and forEach', async () => { - receivedMessages - .filter(msg => msg) - .forEach(msgBody => { - createProcessChildSpan(msgBody); - }); - expectReceiver2ProcessWith1ChildEach(getTestSpans()); - }); - - it.skip('should create processing child with for(msg of messages)', () => { - for (const msg of receivedMessages) { - createProcessChildSpan(msg.Body); - } - expectReceiver2ProcessWith1ChildEach(getTestSpans()); - }); - - it.skip('should create processing child with array.values() for loop', () => { - for (const msg of receivedMessages.values()) { - createProcessChildSpan(msg.Body); - } - expectReceiver2ProcessWith1ChildEach(getTestSpans()); - }); - - it.skip('should create processing child with array.values() for loop and awaits in process', async () => { - for (const msg of receivedMessages.values()) { - await new Promise(resolve => setImmediate(resolve)); - createProcessChildSpan(msg.Body); - } - expectReceiver2ProcessWith1ChildEach(getTestSpans()); - }); - - it('should propagate the context of the receive call in process spans loop', async () => { - receivedMessages.forEach(() => { - expect(context.active().getValue(contextKeyFromTest)).toStrictEqual( - contextValueFromTest - ); - }); - }); - }); - describe('hooks', () => { it('sqsResponseHook for sendMessage should add messaging attributes', async () => { const region = 'us-east-1'; @@ -412,94 +209,6 @@ describe('SQS', () => { expect(span.attributes[SEMATTRS_HTTP_STATUS_CODE]).toEqual(200); }); - it('sqsProcessHook called and add message attribute to span', async () => { - const config = { - sqsProcessHook: ( - span: Span, - sqsProcessInfo: AwsSdkSqsProcessHookInformation - ) => { - span.setAttribute( - 'attribute from sqs process hook', - sqsProcessInfo.message.Body! - ); - }, - }; - - instrumentation.setConfig(config); - - const sqs = new AWS.SQS(); - const res = await sqs - .receiveMessage({ - QueueUrl: 'queue/url/for/unittests', - }) - .promise(); - res.Messages?.map( - message => 'some mapping to create child process spans' - ); - - const processSpans = getTestSpans().filter( - s => - s.attributes[SEMATTRS_MESSAGING_OPERATION] === - MESSAGINGOPERATIONVALUES_PROCESS - ); - expect(processSpans.length).toBe(2); - expect( - processSpans[0].attributes['attribute from sqs process hook'] - ).toBe('msg 1 payload'); - expect( - processSpans[1].attributes['attribute from sqs process hook'] - ).toBe('msg 2 payload'); - }); - - it('sqsProcessHook not set in config', async () => { - const sqs = new AWS.SQS(); - const res = await sqs - .receiveMessage({ - QueueUrl: 'queue/url/for/unittests', - }) - .promise(); - res.Messages?.map( - message => 'some mapping to create child process spans' - ); - const processSpans = getTestSpans().filter( - s => - s.attributes[SEMATTRS_MESSAGING_OPERATION] === - MESSAGINGOPERATIONVALUES_PROCESS - ); - expect(processSpans.length).toBe(2); - }); - - it('sqsProcessHook throws does not fail span', async () => { - const config = { - sqsProcessHook: ( - span: Span, - sqsProcessInfo: AwsSdkSqsProcessHookInformation - ) => { - throw new Error('error from sqsProcessHook hook'); - }, - }; - instrumentation.setConfig(config); - - const sqs = new AWS.SQS(); - const res = await sqs - .receiveMessage({ - QueueUrl: 'queue/url/for/unittests', - }) - .promise(); - res.Messages?.map( - message => 'some mapping to create child process spans' - ); - - const processSpans = getTestSpans().filter( - s => - s.attributes[SEMATTRS_MESSAGING_OPERATION] === - MESSAGINGOPERATIONVALUES_PROCESS - ); - expect(processSpans.length).toBe(2); - expect(processSpans[0].status.code).toStrictEqual(SpanStatusCode.UNSET); - expect(processSpans[1].status.code).toStrictEqual(SpanStatusCode.UNSET); - }); - it('bogus sendMessageBatch input should not crash', async () => { const region = 'us-east-1'; const sqs = new AWS.SQS(); From c9bf62a212f7471c7c428809d9dae7c7c9fc44ec Mon Sep 17 00:00:00 2001 From: Siim Kallas Date: Thu, 18 Jul 2024 22:51:55 +0300 Subject: [PATCH 02/14] add tests --- package-lock.json | 2 - .../package.json | 3 +- .../src/services/sqs.ts | 6 +- .../src/types.ts | 5 -- .../test/sqs.test.ts | 85 ++++++++++++++++--- 5 files changed, 77 insertions(+), 24 deletions(-) diff --git a/package-lock.json b/package-lock.json index 5c23d15b67..dde17e51b1 100644 --- a/package-lock.json +++ b/package-lock.json @@ -38404,7 +38404,6 @@ "dependencies": { "@opentelemetry/core": "^1.8.0", "@opentelemetry/instrumentation": "^0.52.0", - "@opentelemetry/propagation-utils": "^0.30.10", "@opentelemetry/semantic-conventions": "^1.22.0" }, "devDependencies": { @@ -51554,7 +51553,6 @@ "@opentelemetry/contrib-test-utils": "^0.40.0", "@opentelemetry/core": "^1.8.0", "@opentelemetry/instrumentation": "^0.52.0", - "@opentelemetry/propagation-utils": "^0.30.10", "@opentelemetry/sdk-trace-base": "^1.8.0", "@opentelemetry/semantic-conventions": "^1.22.0", "@types/mocha": "8.2.3", diff --git a/plugins/node/opentelemetry-instrumentation-aws-sdk/package.json b/plugins/node/opentelemetry-instrumentation-aws-sdk/package.json index 30c9211053..a18c7fda1e 100644 --- a/plugins/node/opentelemetry-instrumentation-aws-sdk/package.json +++ b/plugins/node/opentelemetry-instrumentation-aws-sdk/package.json @@ -36,7 +36,7 @@ "prewatch": "npm run precompile", "prepublishOnly": "npm run compile", "tdd": "npm run test -- --watch-extensions ts --watch", - "test": "nyc ts-mocha -p tsconfig.json --require '@opentelemetry/contrib-test-utils' 'test/**/sqs.test.ts'", + "test": "nyc ts-mocha -p tsconfig.json --require '@opentelemetry/contrib-test-utils' 'test/**/*.test.ts'", "test-all-versions": "tav", "version:update": "node ../../../scripts/version-update.js", "watch": "tsc -w" @@ -47,7 +47,6 @@ "dependencies": { "@opentelemetry/core": "^1.8.0", "@opentelemetry/instrumentation": "^0.52.0", - "@opentelemetry/propagation-utils": "^0.30.10", "@opentelemetry/semantic-conventions": "^1.22.0" }, "devDependencies": { diff --git a/plugins/node/opentelemetry-instrumentation-aws-sdk/src/services/sqs.ts b/plugins/node/opentelemetry-instrumentation-aws-sdk/src/services/sqs.ts index 64fc9e6e37..74b4fa8f77 100644 --- a/plugins/node/opentelemetry-instrumentation-aws-sdk/src/services/sqs.ts +++ b/plugins/node/opentelemetry-instrumentation-aws-sdk/src/services/sqs.ts @@ -147,8 +147,6 @@ export class SqsServiceExtension implements ServiceExtension { break; case 'ReceiveMessage': { - console.log('ReceiveMessage'); - console.log(span); const messages: SQS.Message[] = response?.data?.Messages || []; span.setAttribute('messaging.batch.message_count', messages.length); @@ -170,8 +168,8 @@ export class SqsServiceExtension implements ServiceExtension { context: spanContext, attributes: { [SEMATTRS_MESSAGING_MESSAGE_ID]: message.MessageId, - } - }) + }, + }); } } break; diff --git a/plugins/node/opentelemetry-instrumentation-aws-sdk/src/types.ts b/plugins/node/opentelemetry-instrumentation-aws-sdk/src/types.ts index 8416b6211d..279e5f513d 100644 --- a/plugins/node/opentelemetry-instrumentation-aws-sdk/src/types.ts +++ b/plugins/node/opentelemetry-instrumentation-aws-sdk/src/types.ts @@ -15,7 +15,6 @@ */ import { Span } from '@opentelemetry/api'; import { InstrumentationConfig } from '@opentelemetry/instrumentation'; -import { SQS } from './aws-sdk.types'; export type CommandInput = Record; @@ -57,10 +56,6 @@ export interface AwsSdkResponseCustomAttributeFunction { (span: Span, responseInfo: AwsSdkResponseHookInformation): void; } -export interface AwsSdkSqsProcessHookInformation { - message: SQS.Message; -} - export type AwsSdkDynamoDBStatementSerializer = ( operation: string, commandInput: CommandInput diff --git a/plugins/node/opentelemetry-instrumentation-aws-sdk/test/sqs.test.ts b/plugins/node/opentelemetry-instrumentation-aws-sdk/test/sqs.test.ts index 066357e134..dad5cd7c2e 100644 --- a/plugins/node/opentelemetry-instrumentation-aws-sdk/test/sqs.test.ts +++ b/plugins/node/opentelemetry-instrumentation-aws-sdk/test/sqs.test.ts @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -import { AwsInstrumentation, AwsSdkSqsProcessHookInformation } from '../src'; +import { AwsInstrumentation } from '../src'; import { getTestSpans, registerInstrumentationTesting, @@ -27,28 +27,19 @@ import type { SQS } from 'aws-sdk'; import { MESSAGINGDESTINATIONKINDVALUES_QUEUE, - MESSAGINGOPERATIONVALUES_RECEIVE, SEMATTRS_HTTP_STATUS_CODE, SEMATTRS_MESSAGING_DESTINATION, SEMATTRS_MESSAGING_DESTINATION_KIND, SEMATTRS_MESSAGING_MESSAGE_ID, - SEMATTRS_MESSAGING_OPERATION, SEMATTRS_MESSAGING_SYSTEM, SEMATTRS_MESSAGING_URL, SEMATTRS_RPC_METHOD, SEMATTRS_RPC_SERVICE, SEMATTRS_RPC_SYSTEM, } from '@opentelemetry/semantic-conventions'; -import { - context, - SpanKind, - SpanStatusCode, - trace, - Span, -} from '@opentelemetry/api'; +import { SpanKind, trace } from '@opentelemetry/api'; import { ReadableSpan } from '@opentelemetry/sdk-trace-base'; import { mockV2AwsSend } from './testing-utils'; -import { Message } from 'aws-sdk/clients/sqs'; import { expect } from 'expect'; import * as sinon from 'sinon'; import * as messageAttributes from '../src/services/MessageAttributes'; @@ -313,4 +304,76 @@ describe('SQS', () => { ); }); }); + + describe('batch receive', () => { + it('adds links to the receive span', async () => { + mockV2AwsSend(responseMockSuccess, { + Messages: [ + { + MessageId: '1', + MessageAttributes: { + traceparent: { + StringValue: + '00-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-aaaaaaaaaaaaaaaa-01', + DataType: 'String', + }, + }, + Body: 'a', + }, + { + MessageId: '2', + MessageAttributes: { + traceparent: { + StringValue: + '00-bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb-bbbbbbbbbbbbbbbb-01', + DataType: 'String', + }, + }, + Body: 'b', + }, + ], + } as AWS.SQS.Types.ReceiveMessageResult); + + const sqs = new AWS.SQS(); + await sqs + .receiveMessage({ + QueueUrl: 'queue/url/for/unittests', + }) + .promise(); + + const [receiveSpan] = getTestSpans(); + const links = receiveSpan.links; + expect(links.length).toBe(2); + expect(links[0].context.traceId).toStrictEqual( + 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' + ); + expect(links[0].context.spanId).toStrictEqual('aaaaaaaaaaaaaaaa'); + expect( + links[0].attributes?.[SEMATTRS_MESSAGING_MESSAGE_ID] + ).toStrictEqual('1'); + expect(links[1].context.traceId).toStrictEqual( + 'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb' + ); + expect(links[1].context.spanId).toStrictEqual('bbbbbbbbbbbbbbbb'); + expect( + links[1].attributes?.[SEMATTRS_MESSAGING_MESSAGE_ID] + ).toStrictEqual('2'); + }); + + it('adds message count to the receive span', async () => { + mockV2AwsSend(responseMockSuccess, { + Messages: [{ Body: 'a' }, { Body: 'b' }, { Body: 'c' }], + } as AWS.SQS.Types.ReceiveMessageResult); + + const sqs = new AWS.SQS(); + await sqs + .receiveMessage({ + QueueUrl: 'queue/url/for/unittests', + }) + .promise(); + + const [receiveSpan] = getTestSpans(); + expect(receiveSpan.attributes['messaging.batch.message_count']).toBe(3); + }); + }); }); From 00214865476199bc88b69c164a928492b9550e92 Mon Sep 17 00:00:00 2001 From: Siim Kallas Date: Thu, 18 Jul 2024 23:05:10 +0300 Subject: [PATCH 03/14] update readme --- plugins/node/opentelemetry-instrumentation-aws-sdk/README.md | 1 - 1 file changed, 1 deletion(-) diff --git a/plugins/node/opentelemetry-instrumentation-aws-sdk/README.md b/plugins/node/opentelemetry-instrumentation-aws-sdk/README.md index a6e01f85ba..df4defeb4e 100644 --- a/plugins/node/opentelemetry-instrumentation-aws-sdk/README.md +++ b/plugins/node/opentelemetry-instrumentation-aws-sdk/README.md @@ -53,7 +53,6 @@ aws-sdk instrumentation has few options available to choose from. You can set th | ----------------------------------------- | ----------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `preRequestHook` | `AwsSdkRequestCustomAttributeFunction` | Hook called before request send, which allow to add custom attributes to span. | | `responseHook` | `AwsSdkResponseCustomAttributeFunction` | Hook for adding custom attributes when response is received from aws. | -| `sqsProcessHook` | `AwsSdkSqsProcessCustomAttributeFunction` | Hook called after starting sqs `process` span (for each sqs received message), which allow to add custom attributes to it. | | `suppressInternalInstrumentation` | `boolean` | Most aws operation use http requests under the hood. Set this to `true` to hide all underlying http spans. | | `sqsExtractContextPropagationFromPayload` | `boolean` | Will parse and extract context propagation headers from SQS Payload, false by default. [When should it be used?](./doc/sns.md#integration-with-sqs) | | `dynamoDBStatementSerializer` | `AwsSdkDynamoDBStatementSerializer` | AWS SDK instrumentation will serialize DynamoDB commands to the `db.statement` attribute using the specified function. Defaults to using a serializer that returns `undefined`. | From 8028d85ba11260e1446a3e4b6f8662905f7ef9dc Mon Sep 17 00:00:00 2001 From: Siim Kallas Date: Fri, 14 Mar 2025 20:06:22 +0200 Subject: [PATCH 04/14] update semconv values --- .../doc/sqs.md | 6 ++-- .../src/semconv.ts | 21 ++++++++++++ .../src/services/sqs.ts | 34 ++++++++----------- .../test/aws-sdk-v3.test.ts | 32 +++++++---------- .../test/sqs.test.ts | 33 ++++++++---------- 5 files changed, 65 insertions(+), 61 deletions(-) create mode 100644 plugins/node/opentelemetry-instrumentation-aws-sdk/src/semconv.ts diff --git a/plugins/node/opentelemetry-instrumentation-aws-sdk/doc/sqs.md b/plugins/node/opentelemetry-instrumentation-aws-sdk/doc/sqs.md index 5631a8a472..70c1171c34 100644 --- a/plugins/node/opentelemetry-instrumentation-aws-sdk/doc/sqs.md +++ b/plugins/node/opentelemetry-instrumentation-aws-sdk/doc/sqs.md @@ -14,10 +14,10 @@ The following methods are automatically enhanced: ### receiveMessage - [Messaging Attributes](https://github.com/open-telemetry/opentelemetry-specification/blob/master/specification/trace/semantic_conventions/messaging.md#messaging-attributes) are added by this instrumentation according to the spec. -- Additional "processing spans" are created for each message received by the application. - If an application invoked `receiveMessage`, and received a 10 messages batch, a single `messaging.operation` = `receive` span will be created for the `receiveMessage` operation, and 10 `messaging.operation` = `process` spans will be created, one for each message. +- Additional "processing spans" are created for each message received by the application. + If an application invoked `receiveMessage`, and received a 10 messages batch, a single `messaging.operation` = `receive` span will be created for the `receiveMessage` operation, and 10 `messaging.operation` = `process` spans will be created, one for each message. Those processing spans are created by the library. This behavior is partially implemented, [See discussion below](#processing-spans). -- Sets the inter process context correctly, so that additional spans created through the process will be linked to parent spans correctly. +- Sets the inter process context correctly, so that additional spans created through the process will be linked to parent spans correctly. This behavior is partially implemented, [See discussion below](#processing-spans). - Extract trace context from SQS MessageAttributes, and set span's `parent` and `links` correctly according to the spec. diff --git a/plugins/node/opentelemetry-instrumentation-aws-sdk/src/semconv.ts b/plugins/node/opentelemetry-instrumentation-aws-sdk/src/semconv.ts new file mode 100644 index 0000000000..b76581242a --- /dev/null +++ b/plugins/node/opentelemetry-instrumentation-aws-sdk/src/semconv.ts @@ -0,0 +1,21 @@ +/* + * Copyright The OpenTelemetry 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 + * + * https://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. + */ + +export const ATTR_MESSAGING_BATCH_MESSAGE_COUNT = + 'messaging.batch.message_count'; +export const ATTR_MESSAGING_DESTINATION_NAME = 'messaging.destination.name'; +export const ATTR_MESSAGING_MESSAGE_ID = 'messaging.message.id'; +export const ATTR_MESSAGING_OPERATION_TYPE = 'messaging.operation.type'; diff --git a/plugins/node/opentelemetry-instrumentation-aws-sdk/src/services/sqs.ts b/plugins/node/opentelemetry-instrumentation-aws-sdk/src/services/sqs.ts index a7d07611b5..794bbd79db 100644 --- a/plugins/node/opentelemetry-instrumentation-aws-sdk/src/services/sqs.ts +++ b/plugins/node/opentelemetry-instrumentation-aws-sdk/src/services/sqs.ts @@ -30,15 +30,15 @@ import { NormalizedResponse, } from '../types'; import { - MESSAGINGDESTINATIONKINDVALUES_QUEUE, - MESSAGINGOPERATIONVALUES_RECEIVE, - SEMATTRS_MESSAGING_DESTINATION, - SEMATTRS_MESSAGING_DESTINATION_KIND, - SEMATTRS_MESSAGING_MESSAGE_ID, - SEMATTRS_MESSAGING_OPERATION, + ATTR_URL_FULL, SEMATTRS_MESSAGING_SYSTEM, - SEMATTRS_MESSAGING_URL, } from '@opentelemetry/semantic-conventions'; +import { + ATTR_MESSAGING_BATCH_MESSAGE_COUNT, + ATTR_MESSAGING_DESTINATION_NAME, + ATTR_MESSAGING_MESSAGE_ID, + ATTR_MESSAGING_OPERATION_TYPE, +} from '../semconv'; import { contextGetter, extractPropagationContext, @@ -57,11 +57,9 @@ export class SqsServiceExtension implements ServiceExtension { let spanName: string | undefined; const spanAttributes: Attributes = { - [SEMATTRS_MESSAGING_SYSTEM]: 'aws.sqs', - [SEMATTRS_MESSAGING_DESTINATION_KIND]: - MESSAGINGDESTINATIONKINDVALUES_QUEUE, - [SEMATTRS_MESSAGING_DESTINATION]: queueName, - [SEMATTRS_MESSAGING_URL]: queueUrl, + [SEMATTRS_MESSAGING_SYSTEM]: 'aws_sqs', + [ATTR_MESSAGING_DESTINATION_NAME]: queueName, + [ATTR_URL_FULL]: queueUrl, }; let isIncoming = false; @@ -72,8 +70,7 @@ export class SqsServiceExtension implements ServiceExtension { isIncoming = true; spanKind = SpanKind.CONSUMER; spanName = `${queueName} receive`; - spanAttributes[SEMATTRS_MESSAGING_OPERATION] = - MESSAGINGOPERATIONVALUES_RECEIVE; + spanAttributes[ATTR_MESSAGING_OPERATION_TYPE] = 'receive'; request.commandInput.MessageAttributeNames = addPropagationFieldsToAttributeNames( @@ -138,10 +135,7 @@ export class SqsServiceExtension implements ServiceExtension { ) => { switch (response.request.commandName) { case 'SendMessage': - span.setAttribute( - SEMATTRS_MESSAGING_MESSAGE_ID, - response?.data?.MessageId - ); + span.setAttribute(ATTR_MESSAGING_MESSAGE_ID, response?.data?.MessageId); break; case 'SendMessageBatch': @@ -151,7 +145,7 @@ export class SqsServiceExtension implements ServiceExtension { case 'ReceiveMessage': { const messages: SQS.Message[] = response?.data?.Messages || []; - span.setAttribute('messaging.batch.message_count', messages.length); + span.setAttribute(ATTR_MESSAGING_BATCH_MESSAGE_COUNT, messages.length); for (const message of messages) { const propagatedContext = propagation.extract( @@ -169,7 +163,7 @@ export class SqsServiceExtension implements ServiceExtension { span.addLink({ context: spanContext, attributes: { - [SEMATTRS_MESSAGING_MESSAGE_ID]: message.MessageId, + [ATTR_MESSAGING_MESSAGE_ID]: message.MessageId, }, }); } diff --git a/plugins/node/opentelemetry-instrumentation-aws-sdk/test/aws-sdk-v3.test.ts b/plugins/node/opentelemetry-instrumentation-aws-sdk/test/aws-sdk-v3.test.ts index 29a1f6bded..a75ab89e9d 100644 --- a/plugins/node/opentelemetry-instrumentation-aws-sdk/test/aws-sdk-v3.test.ts +++ b/plugins/node/opentelemetry-instrumentation-aws-sdk/test/aws-sdk-v3.test.ts @@ -42,12 +42,10 @@ import 'mocha'; import { ReadableSpan } from '@opentelemetry/sdk-trace-base'; import { context, SpanStatusCode, trace, Span } from '@opentelemetry/api'; import { - MESSAGINGDESTINATIONKINDVALUES_QUEUE, + ATTR_URL_FULL, MESSAGINGOPERATIONVALUES_RECEIVE, SEMATTRS_HTTP_STATUS_CODE, SEMATTRS_MESSAGING_DESTINATION, - SEMATTRS_MESSAGING_DESTINATION_KIND, - SEMATTRS_MESSAGING_MESSAGE_ID, SEMATTRS_MESSAGING_OPERATION, SEMATTRS_MESSAGING_SYSTEM, SEMATTRS_MESSAGING_URL, @@ -55,6 +53,10 @@ import { SEMATTRS_RPC_SERVICE, SEMATTRS_RPC_SYSTEM, } from '@opentelemetry/semantic-conventions'; +import { + ATTR_MESSAGING_DESTINATION_NAME, + ATTR_MESSAGING_MESSAGE_ID, +} from '../src/semconv'; import { AttributeNames } from '../src/enums'; import { expect } from 'expect'; import * as fs from 'fs'; @@ -323,17 +325,12 @@ describe('instrumentation-aws-sdk-v3', () => { expect(span.attributes[AttributeNames.AWS_REGION]).toEqual(region); // custom messaging attributes - expect(span.attributes[SEMATTRS_MESSAGING_SYSTEM]).toEqual('aws.sqs'); - expect(span.attributes[SEMATTRS_MESSAGING_DESTINATION_KIND]).toEqual( - MESSAGINGDESTINATIONKINDVALUES_QUEUE - ); - expect(span.attributes[SEMATTRS_MESSAGING_DESTINATION]).toEqual( + expect(span.attributes[SEMATTRS_MESSAGING_SYSTEM]).toEqual('aws_sqs'); + expect(span.attributes[ATTR_MESSAGING_DESTINATION_NAME]).toEqual( 'otel-demo-aws-sdk' ); - expect(span.attributes[SEMATTRS_MESSAGING_URL]).toEqual( - params.QueueUrl - ); - expect(span.attributes[SEMATTRS_MESSAGING_MESSAGE_ID]).toEqual( + expect(span.attributes[ATTR_URL_FULL]).toEqual(params.QueueUrl); + expect(span.attributes[ATTR_MESSAGING_MESSAGE_ID]).toEqual( response.MessageId ); expect(span.attributes[SEMATTRS_HTTP_STATUS_CODE]).toEqual(200); @@ -383,16 +380,11 @@ describe('instrumentation-aws-sdk-v3', () => { expect(span.attributes[AttributeNames.AWS_REGION]).toEqual(region); // messaging semantic attributes - expect(span.attributes[SEMATTRS_MESSAGING_SYSTEM]).toEqual('aws.sqs'); - expect(span.attributes[SEMATTRS_MESSAGING_DESTINATION_KIND]).toEqual( - MESSAGINGDESTINATIONKINDVALUES_QUEUE - ); - expect(span.attributes[SEMATTRS_MESSAGING_DESTINATION]).toEqual( + expect(span.attributes[SEMATTRS_MESSAGING_SYSTEM]).toEqual('aws_sqs'); + expect(span.attributes[ATTR_MESSAGING_DESTINATION_NAME]).toEqual( 'otel-demo-aws-sdk' ); - expect(span.attributes[SEMATTRS_MESSAGING_URL]).toEqual( - params.QueueUrl - ); + expect(span.attributes[ATTR_URL_FULL]).toEqual(params.QueueUrl); expect(span.attributes[SEMATTRS_HTTP_STATUS_CODE]).toEqual(200); }); diff --git a/plugins/node/opentelemetry-instrumentation-aws-sdk/test/sqs.test.ts b/plugins/node/opentelemetry-instrumentation-aws-sdk/test/sqs.test.ts index dad5cd7c2e..55e59ef42d 100644 --- a/plugins/node/opentelemetry-instrumentation-aws-sdk/test/sqs.test.ts +++ b/plugins/node/opentelemetry-instrumentation-aws-sdk/test/sqs.test.ts @@ -26,17 +26,17 @@ import { AWSError } from 'aws-sdk'; import type { SQS } from 'aws-sdk'; import { - MESSAGINGDESTINATIONKINDVALUES_QUEUE, + ATTR_URL_FULL, SEMATTRS_HTTP_STATUS_CODE, - SEMATTRS_MESSAGING_DESTINATION, - SEMATTRS_MESSAGING_DESTINATION_KIND, - SEMATTRS_MESSAGING_MESSAGE_ID, SEMATTRS_MESSAGING_SYSTEM, - SEMATTRS_MESSAGING_URL, SEMATTRS_RPC_METHOD, SEMATTRS_RPC_SERVICE, SEMATTRS_RPC_SYSTEM, } from '@opentelemetry/semantic-conventions'; +import { + ATTR_MESSAGING_DESTINATION_NAME, + ATTR_MESSAGING_MESSAGE_ID, +} from '../src/semconv'; import { SpanKind, trace } from '@opentelemetry/api'; import { ReadableSpan } from '@opentelemetry/sdk-trace-base'; import { mockV2AwsSend } from './testing-utils'; @@ -186,15 +186,12 @@ describe('SQS', () => { expect(span.attributes[AttributeNames.AWS_REGION]).toEqual(region); // custom messaging attributes - expect(span.attributes[SEMATTRS_MESSAGING_SYSTEM]).toEqual('aws.sqs'); - expect(span.attributes[SEMATTRS_MESSAGING_DESTINATION_KIND]).toEqual( - MESSAGINGDESTINATIONKINDVALUES_QUEUE - ); - expect(span.attributes[SEMATTRS_MESSAGING_DESTINATION]).toEqual( + expect(span.attributes[SEMATTRS_MESSAGING_SYSTEM]).toEqual('aws_sqs'); + expect(span.attributes[ATTR_MESSAGING_DESTINATION_NAME]).toEqual( QueueName ); - expect(span.attributes[SEMATTRS_MESSAGING_URL]).toEqual(params.QueueUrl); - expect(span.attributes[SEMATTRS_MESSAGING_MESSAGE_ID]).toEqual( + expect(span.attributes[ATTR_URL_FULL]).toEqual(params.QueueUrl); + expect(span.attributes[ATTR_MESSAGING_MESSAGE_ID]).toEqual( response.MessageId ); expect(span.attributes[SEMATTRS_HTTP_STATUS_CODE]).toEqual(200); @@ -348,16 +345,16 @@ describe('SQS', () => { 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' ); expect(links[0].context.spanId).toStrictEqual('aaaaaaaaaaaaaaaa'); - expect( - links[0].attributes?.[SEMATTRS_MESSAGING_MESSAGE_ID] - ).toStrictEqual('1'); + expect(links[0].attributes?.[ATTR_MESSAGING_MESSAGE_ID]).toStrictEqual( + '1' + ); expect(links[1].context.traceId).toStrictEqual( 'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb' ); expect(links[1].context.spanId).toStrictEqual('bbbbbbbbbbbbbbbb'); - expect( - links[1].attributes?.[SEMATTRS_MESSAGING_MESSAGE_ID] - ).toStrictEqual('2'); + expect(links[1].attributes?.[ATTR_MESSAGING_MESSAGE_ID]).toStrictEqual( + '2' + ); }); it('adds message count to the receive span', async () => { From 4ce0bfbfdbcdcaf0cd77d7945afb42523e898464 Mon Sep 17 00:00:00 2001 From: Siim Kallas Date: Fri, 14 Mar 2025 22:45:31 +0200 Subject: [PATCH 05/14] add tests for v3 --- .../test/aws-sdk-v3.test.ts | 28 +++++++++++++++++-- .../test/sqs.test.ts | 5 +++- 2 files changed, 29 insertions(+), 4 deletions(-) diff --git a/plugins/node/opentelemetry-instrumentation-aws-sdk/test/aws-sdk-v3.test.ts b/plugins/node/opentelemetry-instrumentation-aws-sdk/test/aws-sdk-v3.test.ts index a75ab89e9d..566b2592ef 100644 --- a/plugins/node/opentelemetry-instrumentation-aws-sdk/test/aws-sdk-v3.test.ts +++ b/plugins/node/opentelemetry-instrumentation-aws-sdk/test/aws-sdk-v3.test.ts @@ -32,7 +32,7 @@ import { S3Client, } from '@aws-sdk/client-s3'; import { SQS } from '@aws-sdk/client-sqs'; -import { SpanKind } from '@opentelemetry/api'; +import { propagation, SpanKind } from '@opentelemetry/api'; // set aws environment variables, so tests in non aws environment are able to run process.env.AWS_ACCESS_KEY_ID = 'testing'; @@ -45,15 +45,14 @@ import { ATTR_URL_FULL, MESSAGINGOPERATIONVALUES_RECEIVE, SEMATTRS_HTTP_STATUS_CODE, - SEMATTRS_MESSAGING_DESTINATION, SEMATTRS_MESSAGING_OPERATION, SEMATTRS_MESSAGING_SYSTEM, - SEMATTRS_MESSAGING_URL, SEMATTRS_RPC_METHOD, SEMATTRS_RPC_SERVICE, SEMATTRS_RPC_SYSTEM, } from '@opentelemetry/semantic-conventions'; import { + ATTR_MESSAGING_BATCH_MESSAGE_COUNT, ATTR_MESSAGING_DESTINATION_NAME, ATTR_MESSAGING_MESSAGE_ID, } from '../src/semconv'; @@ -421,6 +420,29 @@ describe('instrumentation-aws-sdk-v3', () => { expect(span.attributes[SEMATTRS_RPC_SERVICE]).toEqual('SQS'); expect(span.attributes[AttributeNames.AWS_REGION]).toEqual(region); expect(span.attributes[SEMATTRS_HTTP_STATUS_CODE]).toEqual(200); + expect(span.attributes[ATTR_MESSAGING_BATCH_MESSAGE_COUNT]).toEqual( + 2 + ); + expect(span.links.length).toBe(2); + + const messages = res.Messages || []; + expect(messages.length).toEqual(span.links.length); + + for (let i = 0; i < span.links.length; i++) { + const link = span.links[i]; + const messageId = messages[i].MessageId; + const traceparent = + messages[i].MessageAttributes?.traceparent.StringValue?.split( + '-' + ) || []; + const traceId = traceparent[1]; + const spanId = traceparent[2]; + expect(link.attributes?.[ATTR_MESSAGING_MESSAGE_ID]).toEqual( + messageId + ); + expect(link.context.traceId).toEqual(traceId); + expect(link.context.spanId).toEqual(spanId); + } done(); }); }); diff --git a/plugins/node/opentelemetry-instrumentation-aws-sdk/test/sqs.test.ts b/plugins/node/opentelemetry-instrumentation-aws-sdk/test/sqs.test.ts index 55e59ef42d..7d1e4aecef 100644 --- a/plugins/node/opentelemetry-instrumentation-aws-sdk/test/sqs.test.ts +++ b/plugins/node/opentelemetry-instrumentation-aws-sdk/test/sqs.test.ts @@ -34,6 +34,7 @@ import { SEMATTRS_RPC_SYSTEM, } from '@opentelemetry/semantic-conventions'; import { + ATTR_MESSAGING_BATCH_MESSAGE_COUNT, ATTR_MESSAGING_DESTINATION_NAME, ATTR_MESSAGING_MESSAGE_ID, } from '../src/semconv'; @@ -370,7 +371,9 @@ describe('SQS', () => { .promise(); const [receiveSpan] = getTestSpans(); - expect(receiveSpan.attributes['messaging.batch.message_count']).toBe(3); + expect(receiveSpan.attributes[ATTR_MESSAGING_BATCH_MESSAGE_COUNT]).toBe( + 3 + ); }); }); }); From fd7a219cb4968b49bcad1eee3df0d69545f683bc Mon Sep 17 00:00:00 2001 From: Siim Kallas Date: Fri, 14 Mar 2025 22:55:28 +0200 Subject: [PATCH 06/14] update sqs docs --- .../doc/sqs.md | 57 ++----------------- 1 file changed, 4 insertions(+), 53 deletions(-) diff --git a/plugins/node/opentelemetry-instrumentation-aws-sdk/doc/sqs.md b/plugins/node/opentelemetry-instrumentation-aws-sdk/doc/sqs.md index 70c1171c34..eca579424b 100644 --- a/plugins/node/opentelemetry-instrumentation-aws-sdk/doc/sqs.md +++ b/plugins/node/opentelemetry-instrumentation-aws-sdk/doc/sqs.md @@ -1,6 +1,6 @@ # SQS -SQS is amazon's managed message queue. Thus, it should follow the [OpenTelemetry specification for Messaging systems](https://github.com/open-telemetry/opentelemetry-specification/blob/master/specification/trace/semantic_conventions/messaging.md). +SQS is Amazon's managed message queue. Thus, it should follow the [OpenTelemetry specification for Messaging systems](https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/). ## Specific trace semantic @@ -8,61 +8,12 @@ The following methods are automatically enhanced: ### sendMessage / sendMessageBatch -- [Messaging Attributes](https://github.com/open-telemetry/opentelemetry-specification/blob/master/specification/trace/semantic_conventions/messaging.md#messaging-attributes) are added by this instrumentation according to the spec. +- [Messaging Attributes](https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/#messaging-attributes) are added by this instrumentation according to the spec. - OpenTelemetry trace context is injected as SQS MessageAttributes, so the service receiving the message can link cascading spans to the trace which created the message. ### receiveMessage -- [Messaging Attributes](https://github.com/open-telemetry/opentelemetry-specification/blob/master/specification/trace/semantic_conventions/messaging.md#messaging-attributes) are added by this instrumentation according to the spec. -- Additional "processing spans" are created for each message received by the application. - If an application invoked `receiveMessage`, and received a 10 messages batch, a single `messaging.operation` = `receive` span will be created for the `receiveMessage` operation, and 10 `messaging.operation` = `process` spans will be created, one for each message. - Those processing spans are created by the library. This behavior is partially implemented, [See discussion below](#processing-spans). +- [Messaging Attributes](https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/#messaging-attributes) are added by this instrumentation according to the spec. - Sets the inter process context correctly, so that additional spans created through the process will be linked to parent spans correctly. - This behavior is partially implemented, [See discussion below](#processing-spans). + When multiple messages are received, the instrumentation will attach spank links to the receiving span containing the trace context and message ID of each message. - Extract trace context from SQS MessageAttributes, and set span's `parent` and `links` correctly according to the spec. - -#### Processing Spans - -See GH issue [here](https://github.com/open-telemetry/opentelemetry-js-contrib/issues/707) - -According to OpenTelemetry specification (and to reasonable expectation for trace structure), user of this library would expect to see one span for the operation of receiving messages batch from SQS, and then, **for each message**, a span with it's own sub-tree for the processing of this specific message. - -For example, if a `receiveMessages` returned 2 messages: - -- `msg1` resulting in storing something to a DB. -- `msg2` resulting in calling an external HTTP endpoint. - -This will result in a creating a DB span that would be the child of `msg1` process span, and an HTTP span that would be the child of `msg2` process span (in opposed to mixing all those operations under the single `receive` span, or start a new trace for each of them). - -Unfortunately, this is not so easy to implement in JS: - -1. The SDK is calling a single callback for the messages batch, and it's not straightforward to understand when each individual message processing starts and ends (and set the context correctly for cascading spans). -2. If async/await is used, context can be lost when returning data from async functions, for example: - -```js -async function asyncRecv() { - const data = await sqs.receiveMessage(recvParams).promise(); - // context of receiveMessage is set here - return data; -} - -async function poll() { - const result = await asyncRecv(); - // context is lost when asyncRecv returns. following spans are created with root context. - await Promise.all( - result.Messages.map((message) => this.processMessage(message)) - ); -} -``` - -Current implementation partially solves this issue by patching the `map` \ `forEach` \ `Filter` functions on the `Messages` array of `receiveMessage` result. This handles issues like the one above, but will not handle situations where the processing is done in other patterns (multiple map\forEach calls, index access to the array, other array operations, etc). This is currently an open issue in the instrumentation. - -User can add custom attributes to the `process` span, by setting a function to `sqsProcessHook` in instrumentation config. For example: - -```js -awsInstrumentationConfig = { - sqsProcessHook: (span, message) => { - span.setAttribute("sqs.receipt_handle", message.params?.ReceiptHandle); - }, -}; -``` From c70fb0e0e93544242af0a519b8fb798dfff4a7c4 Mon Sep 17 00:00:00 2001 From: Siim Kallas Date: Sat, 15 Mar 2025 11:04:36 +0200 Subject: [PATCH 07/14] remove unused dependency --- plugins/node/opentelemetry-instrumentation-aws-sdk/package.json | 1 - 1 file changed, 1 deletion(-) diff --git a/plugins/node/opentelemetry-instrumentation-aws-sdk/package.json b/plugins/node/opentelemetry-instrumentation-aws-sdk/package.json index b0628614ac..74411d5e70 100644 --- a/plugins/node/opentelemetry-instrumentation-aws-sdk/package.json +++ b/plugins/node/opentelemetry-instrumentation-aws-sdk/package.json @@ -46,7 +46,6 @@ "dependencies": { "@opentelemetry/core": "^1.8.0", "@opentelemetry/instrumentation": "^0.57.2", - "@opentelemetry/propagation-utils": "^0.30.16", "@opentelemetry/semantic-conventions": "^1.27.0" }, "devDependencies": { From 19e5cdd5fb2c5eca914e1ed5028f61726849571e Mon Sep 17 00:00:00 2001 From: Siim Kallas Date: Mon, 30 Jun 2025 18:33:01 +0300 Subject: [PATCH 08/14] remove obsolete test file and exports --- .../src/index.ts | 2 - .../test/sqs.test.ts | 383 ------------------ 2 files changed, 385 deletions(-) delete mode 100644 plugins/node/opentelemetry-instrumentation-aws-sdk/test/sqs.test.ts diff --git a/plugins/node/opentelemetry-instrumentation-aws-sdk/src/index.ts b/plugins/node/opentelemetry-instrumentation-aws-sdk/src/index.ts index 410c1c393d..0d7b67b1f8 100644 --- a/plugins/node/opentelemetry-instrumentation-aws-sdk/src/index.ts +++ b/plugins/node/opentelemetry-instrumentation-aws-sdk/src/index.ts @@ -21,8 +21,6 @@ export type { AwsSdkRequestHookInformation, AwsSdkResponseCustomAttributeFunction, AwsSdkResponseHookInformation, - AwsSdkSqsProcessCustomAttributeFunction, - AwsSdkSqsProcessHookInformation, CommandInput, NormalizedRequest, NormalizedResponse, diff --git a/plugins/node/opentelemetry-instrumentation-aws-sdk/test/sqs.test.ts b/plugins/node/opentelemetry-instrumentation-aws-sdk/test/sqs.test.ts deleted file mode 100644 index b03e67ddaa..0000000000 --- a/plugins/node/opentelemetry-instrumentation-aws-sdk/test/sqs.test.ts +++ /dev/null @@ -1,383 +0,0 @@ -/* - * Copyright The OpenTelemetry 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 - * - * https://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 { AwsInstrumentation } from '../src'; -import { - getTestSpans, - registerInstrumentationTesting, -} from '@opentelemetry/contrib-test-utils'; -const instrumentation = registerInstrumentationTesting( - new AwsInstrumentation() -); -import * as AWS from 'aws-sdk'; -import { AWSError } from 'aws-sdk'; -import type { SQS } from 'aws-sdk'; - -import { - ATTR_URL_FULL, - SEMATTRS_HTTP_STATUS_CODE, - SEMATTRS_MESSAGING_SYSTEM, - SEMATTRS_RPC_METHOD, - SEMATTRS_RPC_SERVICE, - SEMATTRS_RPC_SYSTEM, -} from '@opentelemetry/semantic-conventions'; -import { - ATTR_MESSAGING_BATCH_MESSAGE_COUNT, - ATTR_MESSAGING_DESTINATION_NAME, - ATTR_MESSAGING_MESSAGE_ID, -} from '../src/semconv'; -import { SpanKind, trace } from '@opentelemetry/api'; -import { ReadableSpan } from '@opentelemetry/sdk-trace-base'; -import { mockV2AwsSend } from './testing-utils'; -import { expect } from 'expect'; -import * as sinon from 'sinon'; -import * as messageAttributes from '../src/services/MessageAttributes'; -import { AttributeNames } from '../src/enums'; - -// set aws environment variables, so tests in non aws environment are able to run -process.env.AWS_ACCESS_KEY_ID = 'testing'; -process.env.AWS_SECRET_ACCESS_KEY = 'testing'; - -const responseMockSuccess = { - requestId: '0000000000000', - error: null, - httpResponse: { statusCode: 200 }, -} as AWS.Response; - -const extractContextSpy = sinon.spy( - messageAttributes, - 'extractPropagationContext' -); - -describe('SQS', () => { - before(() => { - AWS.config.credentials = { - accessKeyId: 'test key id', - expired: false, - expireTime: new Date(), - secretAccessKey: 'test acc key', - sessionToken: 'test token', - }; - }); - - beforeEach(() => { - mockV2AwsSend(responseMockSuccess, { - Messages: [{ Body: 'msg 1 payload' }, { Body: 'msg 2 payload' }], - } as AWS.SQS.Types.ReceiveMessageResult); - }); - - describe('receive context', () => { - const createReceiveChildSpan = () => { - const childSpan = trace - .getTracerProvider() - .getTracer('default') - .startSpan('child span of SQS.ReceiveMessage'); - childSpan.end(); - }; - - const expectReceiverWithChildSpan = (spans: ReadableSpan[]) => { - const awsReceiveSpan = spans.filter(s => s.kind === SpanKind.CONSUMER); - expect(awsReceiveSpan.length).toBe(1); - const internalSpan = spans.filter(s => s.kind === SpanKind.INTERNAL); - expect(internalSpan.length).toBe(1); - expect(internalSpan[0].parentSpanContext?.spanId).toStrictEqual( - awsReceiveSpan[0].spanContext().spanId - ); - }; - - it('should set parent context in sqs receive callback', done => { - const sqs = new AWS.SQS(); - sqs.receiveMessage( - { - QueueUrl: 'queue/url/for/unittests', - }, - (err: AWSError, data: AWS.SQS.Types.ReceiveMessageResult) => { - expect(err).toBeFalsy(); - createReceiveChildSpan(); - expectReceiverWithChildSpan(getTestSpans()); - done(); - } - ); - }); - - it("should set parent context in sqs receive 'send' callback", done => { - const sqs = new AWS.SQS(); - sqs - .receiveMessage({ - QueueUrl: 'queue/url/for/unittests', - }) - .send((err: AWSError, data: AWS.SQS.Types.ReceiveMessageResult) => { - expect(err).toBeFalsy(); - createReceiveChildSpan(); - expectReceiverWithChildSpan(getTestSpans()); - done(); - }); - }); - - it('should set parent context in sqs receive promise then', async () => { - const sqs = new AWS.SQS(); - await sqs - .receiveMessage({ - QueueUrl: 'queue/url/for/unittests', - }) - .promise() - .then(() => { - createReceiveChildSpan(); - expectReceiverWithChildSpan(getTestSpans()); - }); - }); - - it.skip('should set parent context in sqs receive after await', async () => { - const sqs = new AWS.SQS(); - await sqs - .receiveMessage({ - QueueUrl: 'queue/url/for/unittests', - }) - .promise(); - - createReceiveChildSpan(); - expectReceiverWithChildSpan(getTestSpans()); - }); - - it.skip('should set parent context in sqs receive from async function', async () => { - const asycnReceive = async () => { - const sqs = new AWS.SQS(); - return await sqs - .receiveMessage({ - QueueUrl: 'queue/url/for/unittests', - }) - .promise(); - }; - - await asycnReceive(); - createReceiveChildSpan(); - expectReceiverWithChildSpan(getTestSpans()); - }); - }); - - describe('hooks', () => { - it('sqsResponseHook for sendMessage should add messaging attributes', async () => { - const region = 'us-east-1'; - const sqs = new AWS.SQS(); - sqs.config.update({ region }); - - const QueueName = 'unittest'; - const params = { - QueueUrl: `queue/url/for/${QueueName}`, - MessageBody: 'payload example from v2 without batch', - }; - - const response = await sqs.sendMessage(params).promise(); - - expect(getTestSpans().length).toBe(1); - const [span] = getTestSpans(); - - // make sure we have the general aws attributes: - expect(span.attributes[SEMATTRS_RPC_SYSTEM]).toEqual('aws-api'); - expect(span.attributes[SEMATTRS_RPC_METHOD]).toEqual('SendMessage'); - expect(span.attributes[SEMATTRS_RPC_SERVICE]).toEqual('SQS'); - expect(span.attributes[AttributeNames.AWS_REGION]).toEqual(region); - - // custom messaging attributes - expect(span.attributes[SEMATTRS_MESSAGING_SYSTEM]).toEqual('aws_sqs'); - expect(span.attributes[ATTR_MESSAGING_DESTINATION_NAME]).toEqual( - QueueName - ); - expect(span.attributes[ATTR_URL_FULL]).toEqual(params.QueueUrl); - expect(span.attributes[ATTR_MESSAGING_MESSAGE_ID]).toEqual( - response.MessageId - ); - expect(span.attributes[SEMATTRS_HTTP_STATUS_CODE]).toEqual(200); - }); - - it('bogus sendMessageBatch input should not crash', async () => { - const region = 'us-east-1'; - const sqs = new AWS.SQS(); - sqs.config.update({ region }); - - const QueueName = 'unittest'; - const params = { - QueueUrl: `queue/url/for/${QueueName}`, - Entries: { Key1: { MessageBody: 'This is the first message' } }, - }; - await sqs - .sendMessageBatch(params as unknown as SQS.SendMessageBatchRequest) - .promise(); - - const spans = getTestSpans(); - expect(spans.length).toBe(1); - - // Spot check a single attribute as a sanity check. - expect(spans[0].attributes[SEMATTRS_RPC_METHOD]).toEqual( - 'SendMessageBatch' - ); - }); - }); - - describe('extract payload', () => { - beforeEach(() => { - extractContextSpy.resetHistory(); - }); - it('should not extract from payload even if set', async () => { - mockV2AwsSend(responseMockSuccess, { - Messages: [{ Body: JSON.stringify({ traceparent: 1 }) }], - } as AWS.SQS.Types.ReceiveMessageResult); - - const sqs = new AWS.SQS(); - await sqs - .receiveMessage({ - QueueUrl: 'queue/url/for/unittests1', - }) - .promise(); - expect(extractContextSpy.returnValues[0]?.traceparent).toBeUndefined(); - }); - - it('should extract from payload', async () => { - const traceparent = { - traceparent: { - Value: '00-a1d050b7c8ad93c405e7a0d94cda5b03-23a485dc98b24027-01', - Type: 'String', - }, - }; - instrumentation.setConfig({ - sqsExtractContextPropagationFromPayload: true, - }); - mockV2AwsSend(responseMockSuccess, { - Messages: [ - { Body: JSON.stringify({ MessageAttributes: { traceparent } }) }, - ], - } as AWS.SQS.Types.ReceiveMessageResult); - - const sqs = new AWS.SQS(); - await sqs - .receiveMessage({ - QueueUrl: 'queue/url/for/unittests', - }) - .promise(); - expect(extractContextSpy.returnValues[0]?.traceparent).toStrictEqual( - traceparent - ); - }); - - it('should not extract from payload but from attributes', async () => { - const traceparentInPayload = 'some-trace-parent-value'; - const traceparentInMessageAttributes = { - traceparent: { - StringValue: - '00-a1d050b7c8ad93c405e7a0d94cda5b03-23a485dc98b24027-01', - DataType: 'String', - }, - }; - instrumentation.setConfig({ - sqsExtractContextPropagationFromPayload: false, - }); - mockV2AwsSend(responseMockSuccess, { - Messages: [ - { - MessageAttributes: traceparentInMessageAttributes, - Body: JSON.stringify({ - MessageAttributes: { traceparentInPayload }, - }), - }, - ], - } as AWS.SQS.Types.ReceiveMessageResult); - - const sqs = new AWS.SQS(); - await sqs - .receiveMessage({ - QueueUrl: 'queue/url/for/unittests', - }) - .promise(); - - expect(extractContextSpy.returnValues[0]).toBe( - traceparentInMessageAttributes - ); - }); - }); - - describe('batch receive', () => { - it('adds links to the receive span', async () => { - mockV2AwsSend(responseMockSuccess, { - Messages: [ - { - MessageId: '1', - MessageAttributes: { - traceparent: { - StringValue: - '00-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-aaaaaaaaaaaaaaaa-01', - DataType: 'String', - }, - }, - Body: 'a', - }, - { - MessageId: '2', - MessageAttributes: { - traceparent: { - StringValue: - '00-bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb-bbbbbbbbbbbbbbbb-01', - DataType: 'String', - }, - }, - Body: 'b', - }, - ], - } as AWS.SQS.Types.ReceiveMessageResult); - - const sqs = new AWS.SQS(); - await sqs - .receiveMessage({ - QueueUrl: 'queue/url/for/unittests', - }) - .promise(); - - const [receiveSpan] = getTestSpans(); - const links = receiveSpan.links; - expect(links.length).toBe(2); - expect(links[0].context.traceId).toStrictEqual( - 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' - ); - expect(links[0].context.spanId).toStrictEqual('aaaaaaaaaaaaaaaa'); - expect(links[0].attributes?.[ATTR_MESSAGING_MESSAGE_ID]).toStrictEqual( - '1' - ); - expect(links[1].context.traceId).toStrictEqual( - 'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb' - ); - expect(links[1].context.spanId).toStrictEqual('bbbbbbbbbbbbbbbb'); - expect(links[1].attributes?.[ATTR_MESSAGING_MESSAGE_ID]).toStrictEqual( - '2' - ); - }); - - it('adds message count to the receive span', async () => { - mockV2AwsSend(responseMockSuccess, { - Messages: [{ Body: 'a' }, { Body: 'b' }, { Body: 'c' }], - } as AWS.SQS.Types.ReceiveMessageResult); - - const sqs = new AWS.SQS(); - await sqs - .receiveMessage({ - QueueUrl: 'queue/url/for/unittests', - }) - .promise(); - - const [receiveSpan] = getTestSpans(); - expect(receiveSpan.attributes[ATTR_MESSAGING_BATCH_MESSAGE_COUNT]).toBe( - 3 - ); - }); - }); -}); From ba514c4b6368350cef437eeb3e0a08754074fd49 Mon Sep 17 00:00:00 2001 From: Siim Kallas Date: Mon, 30 Jun 2025 18:47:25 +0300 Subject: [PATCH 09/14] mark semconv values as const --- package-lock.json | 2 -- .../opentelemetry-instrumentation-aws-sdk/package.json | 1 - .../src/semconv.ts | 10 ++++++---- 3 files changed, 6 insertions(+), 7 deletions(-) diff --git a/package-lock.json b/package-lock.json index 5c0c1b102e..2a999d4aea 100644 --- a/package-lock.json +++ b/package-lock.json @@ -32910,7 +32910,6 @@ "dependencies": { "@opentelemetry/core": "^2.0.0", "@opentelemetry/instrumentation": "^0.202.0", - "@opentelemetry/propagation-utils": "^0.31.2", "@opentelemetry/semantic-conventions": "^1.31.0" }, "devDependencies": { @@ -40318,7 +40317,6 @@ "@opentelemetry/contrib-test-utils": "^0.48.0", "@opentelemetry/core": "^2.0.0", "@opentelemetry/instrumentation": "^0.202.0", - "@opentelemetry/propagation-utils": "^0.31.2", "@opentelemetry/sdk-trace-base": "^2.0.0", "@opentelemetry/semantic-conventions": "^1.31.0", "@smithy/node-http-handler": "2.4.0", diff --git a/plugins/node/opentelemetry-instrumentation-aws-sdk/package.json b/plugins/node/opentelemetry-instrumentation-aws-sdk/package.json index ba1a8b2d9a..b98388cc73 100644 --- a/plugins/node/opentelemetry-instrumentation-aws-sdk/package.json +++ b/plugins/node/opentelemetry-instrumentation-aws-sdk/package.json @@ -46,7 +46,6 @@ "dependencies": { "@opentelemetry/core": "^2.0.0", "@opentelemetry/instrumentation": "^0.202.0", - "@opentelemetry/propagation-utils": "^0.31.2", "@opentelemetry/semantic-conventions": "^1.31.0" }, "devDependencies": { diff --git a/plugins/node/opentelemetry-instrumentation-aws-sdk/src/semconv.ts b/plugins/node/opentelemetry-instrumentation-aws-sdk/src/semconv.ts index 9b65605a79..9e1d71de07 100644 --- a/plugins/node/opentelemetry-instrumentation-aws-sdk/src/semconv.ts +++ b/plugins/node/opentelemetry-instrumentation-aws-sdk/src/semconv.ts @@ -160,7 +160,9 @@ export const GEN_AI_TOKEN_TYPE_VALUE_INPUT = 'input' as const; export const GEN_AI_TOKEN_TYPE_VALUE_OUTPUT = 'output' as const; export const ATTR_MESSAGING_BATCH_MESSAGE_COUNT = - 'messaging.batch.message_count'; -export const ATTR_MESSAGING_DESTINATION_NAME = 'messaging.destination.name'; -export const ATTR_MESSAGING_MESSAGE_ID = 'messaging.message.id'; -export const ATTR_MESSAGING_OPERATION_TYPE = 'messaging.operation.type'; + 'messaging.batch.message_count' as const; +export const ATTR_MESSAGING_DESTINATION_NAME = + 'messaging.destination.name' as const; +export const ATTR_MESSAGING_MESSAGE_ID = 'messaging.message.id' as const; +export const ATTR_MESSAGING_OPERATION_TYPE = + 'messaging.operation.type' as const; From 94561e486beb3a4f9e27c9091e8e0ebb0a1a7b88 Mon Sep 17 00:00:00 2001 From: Siim Kallas Date: Thu, 24 Jul 2025 19:37:12 +0300 Subject: [PATCH 10/14] remove unused dep --- package-lock.json | 2 -- packages/instrumentation-aws-sdk/package.json | 1 - 2 files changed, 3 deletions(-) diff --git a/package-lock.json b/package-lock.json index d09d991898..02db753642 100644 --- a/package-lock.json +++ b/package-lock.json @@ -32061,7 +32061,6 @@ "dependencies": { "@opentelemetry/core": "^2.0.0", "@opentelemetry/instrumentation": "^0.203.0", - "@opentelemetry/propagation-utils": "^0.31.3", "@opentelemetry/semantic-conventions": "^1.34.0" }, "devDependencies": { @@ -42305,7 +42304,6 @@ "@opentelemetry/contrib-test-utils": "^0.49.0", "@opentelemetry/core": "^2.0.0", "@opentelemetry/instrumentation": "^0.203.0", - "@opentelemetry/propagation-utils": "^0.31.3", "@opentelemetry/sdk-trace-base": "^2.0.0", "@opentelemetry/semantic-conventions": "^1.34.0", "@smithy/node-http-handler": "2.4.0", diff --git a/packages/instrumentation-aws-sdk/package.json b/packages/instrumentation-aws-sdk/package.json index f5d5623fe0..a868bb1c28 100644 --- a/packages/instrumentation-aws-sdk/package.json +++ b/packages/instrumentation-aws-sdk/package.json @@ -50,7 +50,6 @@ "dependencies": { "@opentelemetry/core": "^2.0.0", "@opentelemetry/instrumentation": "^0.203.0", - "@opentelemetry/propagation-utils": "^0.31.3", "@opentelemetry/semantic-conventions": "^1.34.0" }, "devDependencies": { From 2ae3ff269ac59cf363668e1ccac391eb11d41eea Mon Sep 17 00:00:00 2001 From: Siim Kallas Date: Fri, 25 Jul 2025 11:09:58 +0300 Subject: [PATCH 11/14] ci From 5d50710ad9c6bc598a70bcc0a83853cc929dcb8f Mon Sep 17 00:00:00 2001 From: Trent Mick Date: Tue, 26 Aug 2025 15:28:01 -0700 Subject: [PATCH 12/14] regenerate src/semconv.ts with scripts/gen-semconv-ts.js tool This just re-orders and adds the block comments from the semantic-conventions package. --- .../instrumentation-aws-sdk/src/semconv.ts | 89 +++++++++++++++---- 1 file changed, 71 insertions(+), 18 deletions(-) diff --git a/packages/instrumentation-aws-sdk/src/semconv.ts b/packages/instrumentation-aws-sdk/src/semconv.ts index 5286a90755..b0be6afba8 100644 --- a/packages/instrumentation-aws-sdk/src/semconv.ts +++ b/packages/instrumentation-aws-sdk/src/semconv.ts @@ -20,6 +20,15 @@ * @see https://github.com/open-telemetry/opentelemetry-js/tree/main/semantic-conventions#unstable-semconv */ +/** + * The ARN of the AWS SNS Topic. An Amazon SNS [topic](https://docs.aws.amazon.com/sns/latest/dg/sns-create-topic.html) is a logical access point that acts as a communication channel. + * + * @example arn:aws:sns:us-east-1:123456789012:mystack-mytopic-NZJ5JSMVGFIE + * + * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. + */ +export const ATTR_AWS_SNS_TOPIC_ARN = 'aws.sns.topic.arn' as const; + /** * The name of the operation being performed. * @@ -139,40 +148,84 @@ export const ATTR_GEN_AI_USAGE_INPUT_TOKENS = export const ATTR_GEN_AI_USAGE_OUTPUT_TOKENS = 'gen_ai.usage.output_tokens' as const; +/** + * The number of messages sent, received, or processed in the scope of the batching operation. + * + * @example 0 + * @example 1 + * @example 2 + * + * @note Instrumentations **SHOULD NOT** set `messaging.batch.message_count` on spans that operate with a single message. When a messaging client library supports both batch and single-message API for the same operation, instrumentations **SHOULD** use `messaging.batch.message_count` for batching APIs and **SHOULD NOT** use it for single-message APIs. + * + * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. + */ +export const ATTR_MESSAGING_BATCH_MESSAGE_COUNT = + 'messaging.batch.message_count' as const; + +/** + * The message destination name + * + * @example MyQueue + * @example MyTopic + * + * @note Destination name **SHOULD** uniquely identify a specific queue, topic or other entity within the broker. If + * the broker doesn't have such notion, the destination name **SHOULD** uniquely identify the broker. + * + * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. + */ +export const ATTR_MESSAGING_DESTINATION_NAME = + 'messaging.destination.name' as const; + +/** + * A value used by the messaging system as an identifier for the message, represented as a string. + * + * @example "452a7c7c7c7048c2f887f61572b18fc2" + * + * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. + */ +export const ATTR_MESSAGING_MESSAGE_ID = 'messaging.message.id' as const; + +/** + * A string identifying the type of the messaging operation. + * + * @note If a custom value is used, it **MUST** be of low cardinality. + * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. + */ +export const ATTR_MESSAGING_OPERATION_TYPE = + 'messaging.operation.type' as const; + /** * Enum value "chat" for attribute {@link ATTR_GEN_AI_OPERATION_NAME}. + * + * Chat completion operation such as [OpenAI Chat API](https://platform.openai.com/docs/api-reference/chat) + * + * @experimental This enum value is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. */ export const GEN_AI_OPERATION_NAME_VALUE_CHAT = 'chat' as const; /** * Enum value "aws.bedrock" for attribute {@link ATTR_GEN_AI_SYSTEM}. + * + * AWS Bedrock + * + * @experimental This enum value is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. */ export const GEN_AI_SYSTEM_VALUE_AWS_BEDROCK = 'aws.bedrock' as const; /** * Enum value "input" for attribute {@link ATTR_GEN_AI_TOKEN_TYPE}. + * + * Input tokens (prompt, input, etc.) + * + * @experimental This enum value is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. */ export const GEN_AI_TOKEN_TYPE_VALUE_INPUT = 'input' as const; /** * Enum value "output" for attribute {@link ATTR_GEN_AI_TOKEN_TYPE}. + * + * Output tokens (completion, response, etc.) + * + * @experimental This enum value is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. */ export const GEN_AI_TOKEN_TYPE_VALUE_OUTPUT = 'output' as const; - -export const ATTR_MESSAGING_BATCH_MESSAGE_COUNT = - 'messaging.batch.message_count' as const; -export const ATTR_MESSAGING_DESTINATION_NAME = - 'messaging.destination.name' as const; -export const ATTR_MESSAGING_MESSAGE_ID = 'messaging.message.id' as const; -export const ATTR_MESSAGING_OPERATION_TYPE = - 'messaging.operation.type' as const; - -/** - * Originally from '@opentelemetry/semantic-conventions/incubating' - * https://github.com/open-telemetry/semantic-conventions/blob/main/docs/registry/attributes/aws.md#amazon-sns-attributes - * The ARN of the AWS SNS Topic. An Amazon SNS [topic](https://docs.aws.amazon.com/sns/latest/dg/sns-create-topic.html) - * is a logical access point that acts as a communication channel. - * @example arn:aws:sns:us-east-1:123456789012:mystack-mytopic-NZJ5JSMVGFIE - * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. - */ -export const ATTR_AWS_SNS_TOPIC_ARN = 'aws.sns.topic.arn' as const; From d18c87e7f5e5462b4f853105b2d8b74aced1f543 Mon Sep 17 00:00:00 2001 From: Trent Mick Date: Tue, 26 Aug 2025 15:29:21 -0700 Subject: [PATCH 13/14] fix missed conflict in earlier merge --- packages/instrumentation-aws-sdk/src/types.ts | 6 ------ 1 file changed, 6 deletions(-) diff --git a/packages/instrumentation-aws-sdk/src/types.ts b/packages/instrumentation-aws-sdk/src/types.ts index 69fa570211..649d73fb25 100644 --- a/packages/instrumentation-aws-sdk/src/types.ts +++ b/packages/instrumentation-aws-sdk/src/types.ts @@ -77,18 +77,12 @@ export interface AwsSdkInstrumentationConfig extends InstrumentationConfig { /** hook for adding custom attributes when response is received from aws */ responseHook?: AwsSdkResponseCustomAttributeFunction; -<<<<<<< HEAD -======= /** * Hook for adding custom attributes when exception is received from aws. * This hook is only available with aws sdk v3 */ exceptionHook?: AwsSdkExceptionCustomAttributeFunction; - /** hook for adding custom attribute when an sqs process span is started */ - sqsProcessHook?: AwsSdkSqsProcessCustomAttributeFunction; - ->>>>>>> main /** custom serializer function for the db.statement attribute in DynamoDB spans */ dynamoDBStatementSerializer?: AwsSdkDynamoDBStatementSerializer; From 05104ff1a23982820840ccec9c1f7cfb8b273edc Mon Sep 17 00:00:00 2001 From: Trent Mick Date: Tue, 26 Aug 2025 15:33:59 -0700 Subject: [PATCH 14/14] regenerate src/semconv.ts again (re-ordering and comments) --- .../instrumentation-aws-sdk/src/semconv.ts | 20 +++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/packages/instrumentation-aws-sdk/src/semconv.ts b/packages/instrumentation-aws-sdk/src/semconv.ts index dd3bfacb45..bd23c7775a 100644 --- a/packages/instrumentation-aws-sdk/src/semconv.ts +++ b/packages/instrumentation-aws-sdk/src/semconv.ts @@ -20,6 +20,16 @@ * @see https://github.com/open-telemetry/opentelemetry-js/tree/main/semantic-conventions#unstable-semconv */ +/** + * The ARN of the Secret stored in the Secrets Mangger + * + * @example arn:aws:secretsmanager:us-east-1:123456789012:secret:SecretName-6RandomCharacters + * + * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. + */ +export const ATTR_AWS_SECRETSMANAGER_SECRET_ARN = + 'aws.secretsmanager.secret.arn' as const; + /** * The ARN of the AWS SNS Topic. An Amazon SNS [topic](https://docs.aws.amazon.com/sns/latest/dg/sns-create-topic.html) is a logical access point that acts as a communication channel. * @@ -229,13 +239,3 @@ export const GEN_AI_TOKEN_TYPE_VALUE_INPUT = 'input' as const; * @experimental This enum value is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. */ export const GEN_AI_TOKEN_TYPE_VALUE_OUTPUT = 'output' as const; - -/** - * Originally from '@opentelemetry/semantic-conventions/incubating' - * https://github.com/open-telemetry/semantic-conventions/blob/main/docs/registry/attributes/aws.md#amazon-secrets-manager-attributes - * The ARN of the Secret stored in the Secrets Mangger - * @example arn:aws:secretsmanager:us-east-1:123456789012:secret:SecretName-6RandomCharacters - * @experimental This attribute is experimental and is subject to breaking changes in minor releases of `@opentelemetry/semantic-conventions`. - */ -export const ATTR_AWS_SECRETSMANAGER_SECRET_ARN = - 'aws.secretsmanager.secret.arn' as const;