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,51 @@
 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._connectionsCounter = {
+ idle: 0,
+ pending: 0,
+ used: 0,
+ };
+ 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 +379,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