feat(instrumentation-pg): initial implementation of DB metrics

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

@@ -26,6 +26,8 @@
   Span,
   SpanStatusCode,
   SpanKind,
+  MeterProvider,
+  UpDownCounter,
 } from '@opentelemetry/api';
 import type * as pgTypes from 'pg';
 import type * as pgPoolTypes from 'pg-pool';
@@ -43,8 +45,46 @@
 import { SpanNames } from './enums/SpanNames';
 
 export class PgInstrumentation extends InstrumentationBase<PgInstrumentationConfig> {
+  private _connectionsCount!: UpDownCounter;
+  private _connectionPendingRequests!: UpDownCounter;
+  // Pool events connect, acquire, release and remove can be called
+  // multiple times without changing the values of total, idle and waiting
+  // connections. The _connectionsCounter is used to keep track of latest
+  // values and only update the metrics _connectionsCount and _connectionPendingRequests
+  // when the value change.
+  private _connectionsCounter: utils.poolConnectionsCounter = {
+    used: 0,
+    idle: 0,
+    pending: 0,
+  };
+
   constructor(config: PgInstrumentationConfig = {}) {
     super(PACKAGE_NAME, PACKAGE_VERSION, config);
+    this._setMetricInstruments();
+  }
+
+  override setMeterProvider(meterProvider: MeterProvider) {
+    super.setMeterProvider(meterProvider);
+    this._setMetricInstruments();
+  }
+
+  private _setMetricInstruments() {
+    this._connectionsCount = this.meter.createUpDownCounter(
+      'db.client.connection.count',
+      {
+        description:
+          'The number of connections that are currently in state described by the state attribute.',
+        unit: '{connection}',
+      }
+    );
+    this._connectionPendingRequests = this.meter.createUpDownCounter(
+      'db.client.connection.pending_requests',
+      {
+        description:
+          'The number of current pending requests for an open connection.',
+        unit: '{connection}',
+      }
+    );
   }
 
   protected init() {
@@ -334,6 +374,42 @@
           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
+          );
+        });
+
         if (callback) {
           const parentSpan = trace.getSpan(context.active());
           callback = utils.patchCallbackPGPool(

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

@@ -22,6 +22,7 @@ import {
   Tracer,
   SpanKind,
   diag,
+  UpDownCounter,
 } from '@opentelemetry/api';
 import { AttributeNames } from './enums/AttributeNames';
 import {
@@ -47,6 +48,11 @@ import type * as pgTypes from 'pg';
 import { safeExecuteInTheMiddle } from '@opentelemetry/instrumentation';
 import { SpanNames } from './enums/SpanNames';
 
+// TODO: Replace these constants once a new version of the semantic conventions
+// package is created
+const SEMATTRS_CLIENT_CONNECTION_POOL_NAME = 'db.client.connection.pool.name';
+const SEMATTRS_CLIENT_CONNECTION_STATE = 'db.client.connection.state';
+
 /**
  * Helper function to get a low cardinality span name from whatever info we have
  * about the query.
@@ -258,6 +264,50 @@ export function patchCallback(
   };
 }
 
+export function getPoolName(pool: PgPoolOptionsParams): string {
+  let poolName = '';
+  poolName += (pool?.host ? `${pool.host}` : 'unknown_host') + ':';
+  poolName += (pool?.port ? `${pool.port}` : 'unknown_port') + '/';
+  poolName += pool?.database ? `${pool.database}` : 'unknown_database';
+
+  return poolName.trim();
+}
+
+export interface poolConnectionsCounter {
+  used: number;
+  idle: number;
+  pending: number;
+}
+
+export function updateCounter(
+  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;
+  const used = all - idle;
+
+  connectionCount.add(used - latestCounter.used, {
+    [SEMATTRS_CLIENT_CONNECTION_STATE]: 'used',
+    [SEMATTRS_CLIENT_CONNECTION_POOL_NAME]: poolName,
+  });
+
+  connectionCount.add(idle - latestCounter.idle, {
+    [SEMATTRS_CLIENT_CONNECTION_STATE]: 'idle',
+    [SEMATTRS_CLIENT_CONNECTION_POOL_NAME]: poolName,
+  });
+
+  connectionPendingRequests.add(pending - latestCounter.pending, {
+    [SEMATTRS_CLIENT_CONNECTION_POOL_NAME]: poolName,
+  });
+
+  return { used: used, idle: idle, pending: pending };
+}
+
 export function patchCallbackPGPool(
   span: Span,
   cb: PgPoolCallback