fix(instrumentation-pg): not add duplicate listeners to pg pool

plugins/node/opentelemetry-instrumentation-pg/README.md

@@ -51,6 +51,23 @@ PostgreSQL instrumentation has few options available to choose from. You can set
 | `requireParentSpan` | `boolean` | If true, requires a parent span to create new spans (default false) |
 | `addSqlCommenterCommentToQueries` | `boolean` | If true, adds [sqlcommenter](https://github.com/open-telemetry/opentelemetry-sqlcommenter) specification compliant comment to queries with tracing context (default false). _NOTE: A comment will not be added to queries that already contain `--` or `/* ... */` in them, even if these are not actually part of comments_ |
 
+### Metrics
+
+The value of the environment variable `OTEL_SEMCONV_STABILITY_OPT_IN` (a comma-separated list of values) will define which metrics are being collected.
+Possible values:
+
+- `database` - emit the new database conventions and stop emitting old database conventions.
+- `database/dup` - emit both the old and the stable database conventions, allowing for a seamless transition.
+- The default behavior (in the absence of one of these values) is to continue emitting only the old database conventions.
+
+List of Metrics affected by this environment variable (if the metric is not listed here, it will always be emitted):
+
+| Metric                                  | Emitted with value of `OTEL_SEMCONV_STABILITY_OPT_IN` |
+| --------------------------------------- | ---------------------------- |
+| `db.client.operation.duration`          | `database` or `database/dup` |
+| `db.client.connection.count`            | `database` or `database/dup` |
+| `db.client.connection.pending_requests` | `database` or `database/dup` |
+
 ## Useful links
 
 - For more information on OpenTelemetry, visit: <https://opentelemetry.io/>

plugins/node/opentelemetry-instrumentation-pg/src/instrumentation.ts

@@ -39,6 +39,7 @@
   PostgresCallback,
   PgPoolExtended,
   PgPoolCallback,
+  EVENT_LISTENERS_SET,
 } from './internal-types';
 import { PgInstrumentationConfig } from './types';
 import * as utils from './utils';
@@ -85,6 +86,8 @@
   }
 
   override _updateMetricInstruments() {
+    const emit = utils.getEmitType();
+    if (emit === utils.emitType.OLD_SEM_CONV) return;
     this._operationDuration = this.meter.createHistogram(
       METRIC_DB_CLIENT_OPERATION_DURATION,
       {
@@ -222,6 +225,8 @@
   }
 
   private recordOperationDuration(attributes: Attributes, startTime: HrTime) {
+    const emit = utils.getEmitType();
+    if (emit === utils.emitType.OLD_SEM_CONV) return;
     const metricsAttributes: Attributes = {};
     const keysToCopy = [
       SEMATTRS_DB_SYSTEM,
@@ -435,6 +440,54 @@
     };
   }
 
+  private _setPoolConnectEventListeners(pgPool: PgPoolExtended) {
+    const emit = utils.getEmitType();
+    if (emit === utils.emitType.OLD_SEM_CONV) return;
+    if (pgPool[EVENT_LISTENERS_SET]) return;
+    const poolName = utils.getPoolName(pgPool.options);
+
+    pgPool.on('connect', () => {
+      this._connectionsCounter = utils.updateCounter(
+        poolName,
+        pgPool,
+        this._connectionsCount,
+        this._connectionPendingRequests,
+        this._connectionsCounter
+      );
+    });
+
+    pgPool.on('acquire', () => {
+      this._connectionsCounter = utils.updateCounter(
+        poolName,
+        pgPool,
+        this._connectionsCount,
+        this._connectionPendingRequests,
+        this._connectionsCounter
+      );
+    });
+
+    pgPool.on('remove', () => {
+      this._connectionsCounter = utils.updateCounter(
+        poolName,
+        pgPool,
+        this._connectionsCount,
+        this._connectionPendingRequests,
+        this._connectionsCounter
+      );
+    });
+
+    pgPool.on('release' as any, () => {
+      this._connectionsCounter = utils.updateCounter(
+        poolName,
+        pgPool,
+        this._connectionsCount,
+        this._connectionPendingRequests,
+        this._connectionsCounter
+      );
+    });
+    pgPool[EVENT_LISTENERS_SET] = true;
+  }
+
   private _getPoolConnectPatch() {
     const plugin = this;
     return (originalConnect: typeof pgPoolTypes.prototype.connect) => {
@@ -449,41 +502,7 @@
           attributes: utils.getSemanticAttributesFromPool(this.options),
         });
 
-        this.on('connect', () => {
-          plugin._connectionsCounter = utils.updateCounter(
-            this,
-            plugin._connectionsCount,
-            plugin._connectionPendingRequests,
-            plugin._connectionsCounter
-          );
-        });
-
-        this.on('acquire', () => {
-          plugin._connectionsCounter = utils.updateCounter(
-            this,
-            plugin._connectionsCount,
-            plugin._connectionPendingRequests,
-            plugin._connectionsCounter
-          );
-        });
-
-        this.on('remove', () => {
-          plugin._connectionsCounter = utils.updateCounter(
-            this,
-            plugin._connectionsCount,
-            plugin._connectionPendingRequests,
-            plugin._connectionsCounter
-          );
-        });
-
-        this.on('release' as any, () => {
-          plugin._connectionsCounter = utils.updateCounter(
-            this,
-            plugin._connectionsCount,
-            plugin._connectionPendingRequests,
-            plugin._connectionsCounter
-          );
-        });
+        plugin._setPoolConnectEventListeners(this);
 
         if (callback) {
           const parentSpan = trace.getSpan(context.active());

plugins/node/opentelemetry-instrumentation-pg/src/internal-types.ts

@@ -55,8 +55,13 @@ export interface PgPoolOptionsParams {
   maxClient: number; // maximum size of the pool
 }
 
+export const EVENT_LISTENERS_SET = Symbol(
+  'opentelemetry.instrumentation.pg.eventListenersSet'
+);
+
 export interface PgPoolExtended extends pgPoolTypes<pgTypes.Client> {
   options: PgPoolOptionsParams;
+  [EVENT_LISTENERS_SET]?: boolean; // flag to identify if the event listeners for instrumentation have been set
 }
 
 export type PgClientConnect = (callback?: Function) => Promise<void> | void;

plugins/node/opentelemetry-instrumentation-pg/src/utils.ts

@@ -54,6 +54,22 @@
 import { safeExecuteInTheMiddle } from '@opentelemetry/instrumentation';
 import { SpanNames } from './enums/SpanNames';
 
+export enum emitType {
+  NEW_SEM_CONV,
+  OLD_SEM_CONV,
+  BOTH_SEM_CONV,
+}
+
+export function getEmitType(): emitType {
+  if (process.env.OTEL_SEMCONV_STABILITY_OPT_IN?.includes('database/dup')) {
+    return emitType.BOTH_SEM_CONV;
+  }
+  if (process.env.OTEL_SEMCONV_STABILITY_OPT_IN?.includes('database')) {
+    return emitType.NEW_SEM_CONV;
+  }
+  return emitType.OLD_SEM_CONV;
+}
+
 /**
  * Helper function to get a low cardinality span name from whatever info we have
  * about the query.
@@ -282,12 +298,12 @@
 }
 
 export function updateCounter(
+  poolName: string,
   pool: PgPoolExtended,
   connectionCount: UpDownCounter,
   connectionPendingRequests: UpDownCounter,
   latestCounter: poolConnectionsCounter
 ): poolConnectionsCounter {
-  const poolName = getPoolName(pool.options);
   const all = pool.totalCount;
   const pending = pool.waitingCount;
   const idle = pool.idleCount;