Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 9 additions & 22 deletions package-lock.json

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion packages/instrumentation-amqplib/package.json
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@
"devDependencies": {
"@opentelemetry/api": "^1.3.0",
"@opentelemetry/contrib-test-utils": "^0.49.0",
"@types/amqplib": "^0.5.17",
"@types/amqplib": "^0.10.7",
"@types/lodash": "4.14.199",
"@types/mocha": "10.0.10",
"@types/node": "18.18.14",
Expand Down
19 changes: 7 additions & 12 deletions packages/instrumentation-amqplib/src/amqplib.ts
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,9 @@ import {
CONNECTION_ATTRIBUTES,
getConnectionAttributesFromServer,
getConnectionAttributesFromUrl,
InstrumentationConnection,
InstrumentationConsumeChannel,
InstrumentationConsumeMessage,
InstrumentationMessage,
InstrumentationPublishChannel,
isConfirmChannelTracing,
Expand Down Expand Up @@ -253,18 +255,11 @@ export class AmqplibInstrumentation extends InstrumentationBase<AmqplibInstrumen
this,
url,
socketOptions,
function (this: unknown, err, conn: Connection) {
function (this: unknown, err, conn: InstrumentationConnection) {
if (err == null) {
const urlAttributes = getConnectionAttributesFromUrl(url);
// the type of conn in @types/amqplib is amqp.Connection, but in practice the library send the
// `serverProperties` on the `conn` and not in a property `connection`.
// I don't have capacity to debug it currently but it should probably be fixed in @types or
// in the package itself
// currently setting as any to calm typescript
const serverAttributes = getConnectionAttributesFromServer(
conn as any
);
(conn as any)[CONNECTION_ATTRIBUTES] = {
const serverAttributes = getConnectionAttributesFromServer(conn);
conn[CONNECTION_ATTRIBUTES] = {
...urlAttributes,
...serverAttributes,
};
Expand Down Expand Up @@ -402,7 +397,7 @@ export class AmqplibInstrumentation extends InstrumentationBase<AmqplibInstrumen

const patchedOnMessage = function (
this: unknown,
msg: InstrumentationMessage | null
msg: InstrumentationConsumeMessage | null
) {
// msg is expected to be null for signaling consumer cancel notification
// https://www.rabbitmq.com/consumer-cancel.html
Expand Down Expand Up @@ -724,7 +719,7 @@ export class AmqplibInstrumentation extends InstrumentationBase<AmqplibInstrumen

private callConsumeEndHook(
span: Span,
msg: ConsumeMessage,
msg: InstrumentationMessage,
rejected: boolean | null,
endOperation: EndOperation
) {
Expand Down
12 changes: 9 additions & 3 deletions packages/instrumentation-amqplib/src/utils.ts
Original file line number Diff line number Diff line change
Expand Up @@ -45,12 +45,15 @@ export const CONNECTION_ATTRIBUTES: unique symbol = Symbol(
'opentelemetry.amqplib.connection.attributes'
);

export type InstrumentationConnection = amqp.Connection & {
[CONNECTION_ATTRIBUTES]?: Attributes;
};
export type InstrumentationPublishChannel = (
| amqp.Channel
| amqp.ConfirmChannel
) & { connection: { [CONNECTION_ATTRIBUTES]: Attributes } };
) & { connection: InstrumentationConnection };
export type InstrumentationConsumeChannel = amqp.Channel & {
connection: { [CONNECTION_ATTRIBUTES]: Attributes };
connection: InstrumentationConnection;
[CHANNEL_SPANS_NOT_ENDED]?: {
msg: amqp.ConsumeMessage;
timeOfConsume: HrTime;
Expand All @@ -60,6 +63,9 @@ export type InstrumentationConsumeChannel = amqp.Channel & {
export type InstrumentationMessage = amqp.Message & {
[MESSAGE_STORED_SPAN]?: Span;
};
export type InstrumentationConsumeMessage = amqp.ConsumeMessage & {
[MESSAGE_STORED_SPAN]?: Span;
};

const IS_CONFIRM_CHANNEL_CONTEXT_KEY: symbol = createContextKey(
'opentelemetry.amqplib.channel.is-confirm-channel'
Expand Down Expand Up @@ -117,7 +123,7 @@ const extractConnectionAttributeOrLog = (
};

export const getConnectionAttributesFromServer = (
conn: amqp.Connection['connection']
conn: amqp.Connection
): Attributes => {
const product = conn.serverProperties.product?.toLowerCase?.();
if (product) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ const CHANNEL_CLOSED_IN_TEST = Symbol(
);

describe('amqplib instrumentation promise model', () => {
let conn: amqp.Connection;
let conn: amqp.ChannelModel;
before(async function () {
if (!shouldTest) {
this.skip();
Expand Down
2 changes: 1 addition & 1 deletion packages/instrumentation-amqplib/test/utils.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ import { rabbitMqUrl } from './config';

describe('utils', () => {
describe('getConnectionAttributesFromServer', () => {
let conn: amqp.Connection;
let conn: amqp.ChannelModel;
before(async function () {
if (!shouldTest) {
this.skip();
Expand Down
Loading