Amqplib instrumentation for NestJS won't work correctly

[IonitaCatalin]

Hello there.

As a bit of context, currently working on adding traces on a NestJS application. The prospect of having the amqplib library auto instrumented through the @opentelemetry/instrumentation-amqplib plugin sounds very appealing.

The application consists mostly of a few API aggregates through a gateway and some daemons that listens to a few queue and execute the specified actions.

My main problem with traces for the moment it is that all the instrumentation libraries that are supposed to work inside a single service do work ( spans about function executions for instance are generated) but spans about the queue and the messages are not generated. The @opentelemetry/instrumentation-amqplib seems to be loaded correctly, the opentelemetry logs are not complaining, everything seems to be loading correctly and in working order., but spans from the amqplib are not generated.

The version that I am currently using for the amqplib is 0.10.3.

This is how I bootstrap one of the APIs and also instantiate the OpenTelemetry SDK.

api.ts

import { join } from 'path';

import { NestFactory } from '@nestjs/core';
import { ExpressAdapter, NestExpressApplication } from '@nestjs/platform-express';
import { json, urlencoded } from 'body-parser';
import * as compression from 'compression';
import * as cookieParser from 'cookie-parser';
import { diag, DiagConsoleLogger, DiagLogLevel } from '@opentelemetry/api';
import { AsyncHooksContextManager } from '@opentelemetry/context-async-hooks';
import { CompositePropagator } from '@opentelemetry/core';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';
import { registerInstrumentations } from '@opentelemetry/instrumentation';
import { AmqplibInstrumentation } from '@opentelemetry/instrumentation-amqplib';
import { B3InjectEncoding, B3Propagator } from '@opentelemetry/propagator-b3';
import { JaegerPropagator } from '@opentelemetry/propagator-jaeger';
import { Resource } from '@opentelemetry/resources';
import * as opentelemetry from '@opentelemetry/sdk-node';
import { BatchSpanProcessor } from '@opentelemetry/sdk-trace-base';
import * as express from 'express';
import * as robots from 'express-robots-txt';
import * as helmet from 'helmet';
import * as nocache from 'nocache';
import { AppModule } from './app/app.module';

registerInstrumentations({ instrumentations: [new AmqplibInstrumentation()] });

async function bootstrap(): Promise<void> {
  AppLogger.globalLogLevel = (process.env.ENV_LOG_LEVEL ? parseInt(process.env.ENV_LOG_LEVEL, 10) : null) || AppLogLevel.debug;

  diag.setLogger(new DiagConsoleLogger(), DiagLogLevel.ALL);

  const logger = new AppLogger('api-graphql-route'),
    server = express(),
    exporter = new OTLPTraceExporter();

  void (await new opentelemetry.NodeSDK({
    serviceName: 'api_graphql_route',
    autoDetectResources: false,
    traceExporter: exporter,
    contextManager: new AsyncHooksContextManager().enable(),
    textMapPropagator: new CompositePropagator({
      propagators: [
        new JaegerPropagator(),
        new JaegerPropagator(),
        new B3Propagator(),
        new B3Propagator({
          injectEncoding: B3InjectEncoding.MULTI_HEADER,
        }),
      ],
    }),
    resource: new Resource({
      service_version: process.env.npm_package_version,
    }),
    spanProcessor: new BatchSpanProcessor(exporter, {}),
  })
    .start()
    .then(() => {
      logger.info('OpenTelemetry Node SDK started successfully!');
    }));

  const app = await NestFactory.create<NestExpressApplication>(AppModule, new ExpressAdapter(server), {
    cors: true,
    logger,
  });

  app.use(compression({}));
  app.use(
    helmet({
      contentSecurityPolicy: ParseUtil.parseBoolean(process.env.ENV_PLAYGROUNG) ? false : undefined,
    }),
  );
  app.use(nocache());
  app.use(
    helmet.hsts({
      maxAge: 15768000,
      includeSubDomains: true,
    }),
  );
  app.use(robots({ UserAgent: '*', Disallow: '/' }));
  app.use(cookieParser());
  app.use(json({ limit: '50mb' }));
  app.use(
    urlencoded({
      limit: '50mb',
      extended: true,
    }),
  );

  const port = 4005;

  await app.listen(port, () => {
    logger.info(`Listening at http://localhost:${port}`);
  });

  process.on('unhandledRejection', (err: any): void => {
    logger.error(err.message);
    logger.error(JSON.stringify(err) || err);
  });
}

void bootstrap();

This is how I bootstrap a Daemon type application, the same principle applies.

daemon.ts

import { NestFactory } from '@nestjs/core';
import { CustomTransportStrategy } from '@nestjs/microservices';
import { ExpressAdapter, NestExpressApplication } from '@nestjs/platform-express';
import { diag, DiagConsoleLogger, DiagLogLevel } from '@opentelemetry/api';
import { AsyncHooksContextManager } from '@opentelemetry/context-async-hooks';
import { CompositePropagator } from '@opentelemetry/core';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';
import { registerInstrumentations } from '@opentelemetry/instrumentation';
import { AmqplibInstrumentation } from '@opentelemetry/instrumentation-amqplib';
import { B3InjectEncoding, B3Propagator } from '@opentelemetry/propagator-b3';
import { JaegerPropagator } from '@opentelemetry/propagator-jaeger';
import { Resource } from '@opentelemetry/resources';
import * as opentelemetry from '@opentelemetry/sdk-node';
import { BatchSpanProcessor } from '@opentelemetry/sdk-trace-base';
import * as express from 'express';
import { DaemonExceptionFilter } from '../exceptions';

registerInstrumentations({ instrumentations: [new AmqplibInstrumentation()] });

export async function bootstrap(name: string, moduleClass: NestExpressApplication, strategy: CustomTransportStrategy, port: number): Promise<void> {
  diag.setLogger(new DiagConsoleLogger(), DiagLogLevel.ALL);

  const logger = new AppLogger(name),
    server = express(),
    exporter = new OTLPTraceExporter();

  void (await new opentelemetry.NodeSDK({
    serviceName: 'daemon_routing',
    autoDetectResources: false,
    traceExporter: exporter,
    contextManager: new AsyncHooksContextManager().enable(),
    textMapPropagator: new CompositePropagator({
      propagators: [
        new JaegerPropagator(),
        new JaegerPropagator(),
        new B3Propagator(),
        new B3Propagator({
          injectEncoding: B3InjectEncoding.MULTI_HEADER,
        }),
      ],
    }),
    resource: new Resource({
      service_version: process.env.npm_package_version,
    }),
    spanProcessor: new BatchSpanProcessor(exporter, {}),
  })
    .start()
    .then(() => {
      logger.info('OpenTelemetry Node SDK started successfully!');
    }));

  const application = await NestFactory.create<NestExpressApplication>(moduleClass, new ExpressAdapter(server), {
    cors: true,
    logger,
  });

  application.useGlobalFilters(new DaemonExceptionFilter());

  application.connectMicroservice({
    strategy,
  });
  await application.startAllMicroservices();

  await application.listen(port, () => {
    logger.info(`Listening at http://localhost:${port}`);
  });
}

  await bootstrap(
    'Daemon',
    DaemonModule,
    new RabbitTransportStrategy(QUEUE, 'daemon'),
    9100,
  );

They are pretty identical although the daemon uses the @nestjs/microservices and also a custom RabbitMQTransportStrategy

Could it possibly be due to using a Strategy from NestJS that's causing the instrumentation library to malfunction, or probably a version mismatch. I am pretty much at a lost currently.

If anyone had this issue before and it's available to help, I would be very grateful.

I am not confident enough to say this is a general problem with the instrumentation library or just something that I am simply doing wrong.

• [ ] This only affects the JavaScript OpenTelemetry library
• [ ] This may affect other libraries, but I would like to get opinions here first

[github-actions[bot]]

This issue is stale because it has been open 60 days with no activity. Remove stale label or comment or this will be closed in 14 days.

[github-actions[bot]]

This issue was closed because it has been stale for 14 days with no activity.

[MaksimKiselev]

@IonitaCatalin hey.

I got same issue with similar setup.

Spent much time and finally found the reason: in my second service import of otelSDK (which instance of NodeSDK from tracing.ts) was be not on the first line.

It's very important to have import of otelSDK/NodeSDK on the first line and otelSDK.start() on the first line in bootstrap method.

[Endiaoshuai]

@MaksimKiselev placing "import { otelSDK } from "./tracing";" at the first line of the main.ts file has resolved my problem.