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

plugins/node/instrumentation-amqplib/package.json

@@ -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",

plugins/node/instrumentation-amqplib/src/amqplib.ts

@@ -30,7 +30,6 @@ import {
 } from '@opentelemetry/core';
 import {
   InstrumentationBase,
-  InstrumentationModuleDefinition,
   InstrumentationNodeModuleDefinition,
   InstrumentationNodeModuleFile,
   isWrapped,
@@ -41,11 +40,15 @@ import {
   MessagingOperationValues,
   MessagingDestinationKindValues,
 } from '@opentelemetry/semantic-conventions';
-import type * as amqp from 'amqplib';
 import {
   AmqplibInstrumentationConfig,
+  Connection,
+  ConsumeMessage,
   DEFAULT_CONFIG,
   EndOperation,
+  Message,
+  Options,
+  Replies,
 } from './types';
 import {
   CHANNEL_CONSUME_TIMEOUT_TIMER,
@@ -64,7 +67,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) {
@@ -79,31 +82,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,
@@ -231,22 +232,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
@@ -321,7 +322,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 {
@@ -330,10 +331,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
@@ -375,9 +376,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)
@@ -431,7 +432,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);
               }
@@ -475,8 +476,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(
@@ -510,7 +511,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);
@@ -572,7 +573,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
@@ -623,7 +624,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);
 
@@ -689,25 +690,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);
         }

plugins/node/instrumentation-amqplib/src/internal-types.ts

@@ -0,0 +1,15 @@
+/*
+ * 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.
+ */