Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix(instrumentation-amqplib): move @types/amqplib into dev deps #1320

Merged
merged 12 commits into from
Feb 7, 2023
Merged

fix(instrumentation-amqplib): move @types/amqplib into dev deps #1320

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
54266b8
fix(instrumentation-amqplib): move `@types/amqplib` into dev deps
AbhiPrasad Dec 6, 2022
0aac520
Merge branch 'main' into abhi-types-amqplib
AbhiPrasad Jan 9, 2023
b66d179
refactor(instrumentation-amqplib): vendor in types into internal-types
AbhiPrasad Jan 9, 2023
d7b573a
refactor(instrumentation-amqplib): revert prettier formatting
AbhiPrasad Jan 9, 2023
89d1731
Merge remote-tracking branch 'otel/main' into abhi-types-amqplib
AbhiPrasad Jan 16, 2023
657f169
refactor(instrumentation-amqplib): add code comments for vendoring
AbhiPrasad Jan 16, 2023
104ff7e
refactor(instrumentation-amqplib): delete unused file + run lint fixer
AbhiPrasad Jan 16, 2023
75566b7
refactor(instrumentation-amqplib): fix ts eslint errorsOC
AbhiPrasad Jan 16, 2023
c684f93
refactor(instrumentation-amqplib): fix lint
AbhiPrasad Jan 16, 2023
2f42384
Merge branch 'main' into abhi-types-amqplib
blumamir Feb 7, 2023
10bf81d
Merge branch 'main' into abhi-types-amqplib
blumamir Feb 7, 2023
8353d80
Merge branch 'main' into abhi-types-amqplib
blumamir Feb 7, 2023
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
4 changes: 2 additions & 2 deletions plugins/node/instrumentation-amqplib/package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,12 +48,12 @@
"dependencies": {
"@opentelemetry/core": "^1.8.0",
"@opentelemetry/instrumentation": "^0.34.0",
"@opentelemetry/semantic-conventions": "^1.0.0",
"@types/amqplib": "^0.5.17"
"@opentelemetry/semantic-conventions": "^1.0.0"
},
"devDependencies": {
"@opentelemetry/api": "^1.3.0",
"@opentelemetry/contrib-test-utils": "^0.33.0",
"@types/amqplib": "^0.5.17",
"@types/lodash": "4.14.178",
"@types/mocha": "8.2.3",
"@types/sinon": "10.0.2",
Expand Down
85 changes: 44 additions & 41 deletions plugins/node/instrumentation-amqplib/src/amqplib.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,6 @@ import {
} from '@opentelemetry/core';
import {
InstrumentationBase,
InstrumentationModuleDefinition,
InstrumentationNodeModuleDefinition,
InstrumentationNodeModuleFile,
isWrapped,
Expand All @@ -41,7 +40,13 @@ import {
MessagingOperationValues,
MessagingDestinationKindValues,
} from '@opentelemetry/semantic-conventions';
import type * as amqp from 'amqplib';
import {
Connection,
ConsumeMessage,
Message,
Options,
Replies,
} from './internal-types';
import {
AmqplibInstrumentationConfig,
DEFAULT_CONFIG,
Expand All @@ -64,7 +69,7 @@ import {
} from './utils';
import { VERSION } from './version';

export class AmqplibInstrumentation extends InstrumentationBase<typeof amqp> {
export class AmqplibInstrumentation extends InstrumentationBase {
protected override _config!: AmqplibInstrumentationConfig;

constructor(config?: AmqplibInstrumentationConfig) {
Expand All @@ -79,31 +84,29 @@ export class AmqplibInstrumentation extends InstrumentationBase<typeof amqp> {
this._config = Object.assign({}, DEFAULT_CONFIG, config);
}

protected init(): InstrumentationModuleDefinition<typeof amqp> {
const channelModelModuleFile =
new InstrumentationNodeModuleFile<amqp.Channel>(
'amqplib/lib/channel_model.js',
['>=0.5.5'],
this.patchChannelModel.bind(this),
this.unpatchChannelModel.bind(this)
);
protected init() {
const channelModelModuleFile = new InstrumentationNodeModuleFile(
'amqplib/lib/channel_model.js',
['>=0.5.5'],
this.patchChannelModel.bind(this),
this.unpatchChannelModel.bind(this)
);

const callbackModelModuleFile =
new InstrumentationNodeModuleFile<amqp.Channel>(
'amqplib/lib/callback_model.js',
['>=0.5.5'],
this.patchChannelModel.bind(this),
this.unpatchChannelModel.bind(this)
);
const callbackModelModuleFile = new InstrumentationNodeModuleFile(
'amqplib/lib/callback_model.js',
['>=0.5.5'],
this.patchChannelModel.bind(this),
this.unpatchChannelModel.bind(this)
);

const connectModuleFile = new InstrumentationNodeModuleFile<amqp.Channel>(
const connectModuleFile = new InstrumentationNodeModuleFile(
'amqplib/lib/connect.js',
['>=0.5.5'],
this.patchConnect.bind(this),
this.unpatchConnect.bind(this)
);

const module = new InstrumentationNodeModuleDefinition<typeof amqp>(
const module = new InstrumentationNodeModuleDefinition(
'amqplib',
['>=0.5.5'],
undefined,
Expand Down Expand Up @@ -231,22 +234,22 @@ export class AmqplibInstrumentation extends InstrumentationBase<typeof amqp> {

private getConnectPatch(
original: (
url: string | amqp.Options.Connect,
url: string | Options.Connect,
socketOptions: any,
openCallback: (err: any, connection: amqp.Connection) => void
) => amqp.Connection
openCallback: (err: any, connection: Connection) => void
) => Connection
) {
return function patchedConnect(
this: unknown,
url: string | amqp.Options.Connect,
url: string | Options.Connect,
socketOptions: any,
openCallback: Function
) {
return original.call(
this,
url,
socketOptions,
function (this: unknown, err, conn: amqp.Connection) {
function (this: unknown, err, conn: Connection) {
if (err == null) {
const urlAttributes = getConnectionAttributesFromUrl(url);
// the type of conn in @types/amqplib is amqp.Connection, but in practice the library send the
Expand Down Expand Up @@ -321,7 +324,7 @@ export class AmqplibInstrumentation extends InstrumentationBase<typeof amqp> {
const self = this;
return function ack(
this: InstrumentationConsumeChannel,
message: amqp.Message,
message: Message,
allUpToOrRequeue?: boolean,
requeue?: boolean
): void {
Expand All @@ -330,10 +333,10 @@ export class AmqplibInstrumentation extends InstrumentationBase<typeof amqp> {
const requeueResolved =
endOperation === EndOperation.Reject ? allUpToOrRequeue : requeue;

const spansNotEnded: { msg: amqp.Message }[] =
const spansNotEnded: { msg: Message }[] =
channel[CHANNEL_SPANS_NOT_ENDED] ?? [];
const msgIndex = spansNotEnded.findIndex(
msgDetails => msgDetails.msg === message
(msgDetails) => msgDetails.msg === message
);
if (msgIndex < 0) {
// should not happen in happy flow
Expand Down Expand Up @@ -375,9 +378,9 @@ export class AmqplibInstrumentation extends InstrumentationBase<typeof amqp> {
return function consume(
this: InstrumentationConsumeChannel,
queue: string,
onMessage: (msg: amqp.ConsumeMessage | null) => void,
options?: amqp.Options.Consume
): Promise<amqp.Replies.Consume> {
onMessage: (msg: ConsumeMessage | null) => void,
options?: Options.Consume
): Promise<Replies.Consume> {
const channel = this;
if (
!Object.prototype.hasOwnProperty.call(channel, CHANNEL_SPANS_NOT_ENDED)
Expand Down Expand Up @@ -431,7 +434,7 @@ export class AmqplibInstrumentation extends InstrumentationBase<typeof amqp> {
if (self._config.consumeHook) {
safeExecuteInTheMiddle(
() => self._config.consumeHook!(span, { moduleVersion, msg }),
e => {
(e) => {
if (e) {
diag.error('amqplib instrumentation: consumerHook error', e);
}
Expand Down Expand Up @@ -475,8 +478,8 @@ export class AmqplibInstrumentation extends InstrumentationBase<typeof amqp> {
exchange: string,
routingKey: string,
content: Buffer,
options?: amqp.Options.Publish,
callback?: (err: any, ok: amqp.Replies.Empty) => void
options?: Options.Publish,
callback?: (err: any, ok: Replies.Empty) => void
): boolean {
const channel = this;
const { span, modifiedOptions } = self.createPublishSpan(
Expand Down Expand Up @@ -510,7 +513,7 @@ export class AmqplibInstrumentation extends InstrumentationBase<typeof amqp> {
const patchedOnConfirm = function (
this: unknown,
err: any,
ok: amqp.Replies.Empty
ok: Replies.Empty
) {
try {
callback?.call(this, err, ok);
Expand Down Expand Up @@ -572,7 +575,7 @@ export class AmqplibInstrumentation extends InstrumentationBase<typeof amqp> {
exchange: string,
routingKey: string,
content: Buffer,
options?: amqp.Options.Publish
options?: Options.Publish
): boolean {
if (isConfirmChannelTracing(context.active())) {
// work already done
Expand Down Expand Up @@ -623,7 +626,7 @@ export class AmqplibInstrumentation extends InstrumentationBase<typeof amqp> {
exchange: string,
routingKey: string,
channel: InstrumentationPublishChannel,
options?: amqp.Options.Publish
options?: Options.Publish
) {
const normalizedExchange = normalizeExchange(exchange);

Expand Down Expand Up @@ -689,25 +692,25 @@ export class AmqplibInstrumentation extends InstrumentationBase<typeof amqp> {
operation: EndOperation,
requeue: boolean | undefined
) {
const spansNotEnded: { msg: amqp.Message }[] =
const spansNotEnded: { msg: Message }[] =
channel[CHANNEL_SPANS_NOT_ENDED] ?? [];
spansNotEnded.forEach(msgDetails => {
spansNotEnded.forEach((msgDetails) => {
this.endConsumerSpan(msgDetails.msg, isRejected, operation, requeue);
});
channel[CHANNEL_SPANS_NOT_ENDED] = [];
}

private callConsumeEndHook(
span: Span,
msg: amqp.ConsumeMessage,
msg: ConsumeMessage,
rejected: boolean | null,
endOperation: EndOperation
) {
if (!this._config.consumeEndHook) return;

safeExecuteInTheMiddle(
() => this._config.consumeEndHook!(span, { msg, rejected, endOperation }),
e => {
(e) => {
if (e) {
diag.error('amqplib instrumentation: consumerEndHook error', e);
}
Expand Down
Loading