feat(instrumentation-pg): implementation of metric operation duration

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

@@ -19,13 +19,16 @@
   InstrumentationNodeModuleDefinition,
   safeExecuteInTheMiddle,
 } from '@opentelemetry/instrumentation';
-
 import {
   context,
   trace,
   Span,
   SpanStatusCode,
   SpanKind,
+  Histogram,
+  ValueType,
+  Attributes,
+  HrTime,
 } from '@opentelemetry/api';
 import type * as pgTypes from 'pg';
 import type * as pgPoolTypes from 'pg-pool';
@@ -41,12 +44,45 @@
 import { addSqlCommenterComment } from '@opentelemetry/sql-common';
 import { PACKAGE_NAME, PACKAGE_VERSION } from './version';
 import { SpanNames } from './enums/SpanNames';
+import {
+  hrTime,
+  hrTimeDuration,
+  hrTimeToMilliseconds,
+} from '@opentelemetry/core';
+import {
+  DBSYSTEMVALUES_POSTGRESQL,
+  SEMATTRS_DB_SYSTEM,
+  ATTR_ERROR_TYPE,
+  ATTR_SERVER_PORT,
+  ATTR_SERVER_ADDRESS,
+} from '@opentelemetry/semantic-conventions';
+import {
+  METRIC_DB_CLIENT_OPERATION_DURATION,
+  ATTR_DB_NAMESPACE,
+} from '@opentelemetry/semantic-conventions/incubating';
 
 export class PgInstrumentation extends InstrumentationBase<PgInstrumentationConfig> {
+  private _operationDuration!: Histogram;
   constructor(config: PgInstrumentationConfig = {}) {
     super(PACKAGE_NAME, PACKAGE_VERSION, config);
   }
 
+  override _updateMetricInstruments() {
+    this._operationDuration = this.meter.createHistogram(
+      METRIC_DB_CLIENT_OPERATION_DURATION,
+      {
+        description: 'Duration of database client operations.',
+        unit: 's',
+        valueType: ValueType.DOUBLE,
+        advice: {
+          explicitBucketBoundaries: [
+            0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1, 5, 10,
+          ],
+        },
+      }
+    );
+  }
+
   protected init() {
     const modulePG = new InstrumentationNodeModuleDefinition(
       'pg',
@@ -146,6 +182,27 @@
     };
   }
 
+  private recordOperationDuration(attributes: Attributes, startTime: HrTime) {
+    const metricsAttributes: Attributes = {};
+    const keysToCopy = [
+      SEMATTRS_DB_SYSTEM,
+      ATTR_DB_NAMESPACE,
+      ATTR_ERROR_TYPE,
+      ATTR_SERVER_PORT,
+      ATTR_SERVER_ADDRESS,
+    ];
+
+    keysToCopy.forEach(key => {
+      if (key in attributes) {
+        metricsAttributes[key] = attributes[key];
+      }
+    });
+
+    const durationSeconds =
+      hrTimeToMilliseconds(hrTimeDuration(startTime, hrTime())) / 1000;
+    this._operationDuration.record(durationSeconds, metricsAttributes);
+  }
+
   private _getClientQueryPatch() {
     const plugin = this;
     return (original: typeof pgTypes.Client.prototype.query) => {
@@ -154,6 +211,7 @@
         if (utils.shouldSkipInstrumentation(plugin.getConfig())) {
           return original.apply(this, args as never);
         }
+        const startTime = hrTime();
 
         // client.query(text, cb?), client.query(text, values, cb?), and
         // client.query(configObj, cb?) are all valid signatures. We construct
@@ -179,6 +237,17 @@
           ? (arg0 as utils.ObjectWithText)
           : undefined;
 
+        const attributes: Attributes = {
+          [SEMATTRS_DB_SYSTEM]: DBSYSTEMVALUES_POSTGRESQL,
+          [ATTR_DB_NAMESPACE]: this.database,
+          [ATTR_SERVER_PORT]: this.connectionParameters.port,
+          [ATTR_SERVER_ADDRESS]: this.connectionParameters.host,
+        };
+
+        const recordDuration = () => {
+          plugin.recordOperationDuration(attributes, startTime);
+        };
+
         const instrumentationConfig = plugin.getConfig();
 
         const span = utils.handleConfigQuery.call(
@@ -209,7 +278,9 @@
             args[args.length - 1] = utils.patchCallback(
               instrumentationConfig,
               span,
-              args[args.length - 1] as PostgresCallback // nb: not type safe.
+              args[args.length - 1] as PostgresCallback, // nb: not type safe.
+              attributes,
+              recordDuration
             );
 
             // If a parent span exists, bind the callback
@@ -224,7 +295,9 @@
             let callback = utils.patchCallback(
               plugin.getConfig(),
               span,
-              queryConfig.callback as PostgresCallback // nb: not type safe.
+              queryConfig.callback as PostgresCallback, // nb: not type safe.
+              attributes,
+              recordDuration
             );
 
             // If a parent span existed, bind the callback
@@ -282,12 +355,12 @@
         try {
           result = original.apply(this, args as never);
         } catch (e: unknown) {
-          // span.recordException(e);
           span.setStatus({
             code: SpanStatusCode.ERROR,
             message: utils.getErrorMessage(e),
           });
           span.end();
+          attributes[ATTR_ERROR_TYPE] = utils.getErrorMessage(e);
           throw e;
         }
 
@@ -309,6 +382,7 @@
                   message: error.message,
                 });
                 span.end();
+                attributes[ATTR_ERROR_TYPE] = utils.getErrorMessage(error);
                 reject(error);
               });
             });

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

@@ -22,6 +22,7 @@ import {
   Tracer,
   SpanKind,
   diag,
+  Attributes,
 } from '@opentelemetry/api';
 import { AttributeNames } from './enums/AttributeNames';
 import {
@@ -236,15 +237,17 @@ export function handleExecutionResult(
 export function patchCallback(
   instrumentationConfig: PgInstrumentationConfig,
   span: Span,
-  cb: PostgresCallback
+  cb: PostgresCallback,
+  attributes: Attributes,
+  recordDuration: { (): void }
 ): PostgresCallback {
   return function patchedCallback(
     this: PgClientExtended,
     err: Error,
     res: object
   ) {
     if (err) {
-      // span.recordException(err);
+      attributes['error.type'] = getErrorMessage(err);
       span.setStatus({
         code: SpanStatusCode.ERROR,
         message: err.message,
@@ -253,6 +256,7 @@ export function patchCallback(
       handleExecutionResult(instrumentationConfig, span, res);
     }
 
+    recordDuration();
     span.end();
     cb.call(this, err, res);
   };

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

@@ -30,6 +30,7 @@ import {
   InMemorySpanExporter,
   SimpleSpanProcessor,
 } from '@opentelemetry/sdk-trace-base';
+import { DataPoint, Histogram } from '@opentelemetry/sdk-metrics';
 import * as assert from 'assert';
 import type * as pg from 'pg';
 import * as sinon from 'sinon';
@@ -50,6 +51,7 @@ import {
   SEMATTRS_NET_PEER_PORT,
   SEMATTRS_DB_USER,
   DBSYSTEMVALUES_POSTGRESQL,
+  ATTR_ERROR_TYPE,
 } from '@opentelemetry/semantic-conventions';
 import { addSqlCommenterComment } from '@opentelemetry/sql-common';
 
@@ -960,4 +962,102 @@ describe('pg', () => {
       });
     });
   });
+
+  describe('pg metrics', () => {
+    let metricReader: testUtils.TestMetricReader;
+
+    beforeEach(() => {
+      metricReader = testUtils.initMeterProvider(instrumentation);
+    });
+
+    it('should generate db.client.operation.duration metric', done => {
+      client.query('SELECT NOW()', async (_, ret) => {
+        assert.ok(ret, 'query should be executed');
+
+        const { resourceMetrics, errors } = await metricReader.collect();
+        assert.deepEqual(
+          errors,
+          [],
+          'expected no errors from the callback during metric collection'
+        );
+
+        const metrics = resourceMetrics.scopeMetrics[0].metrics;
+        assert.strictEqual(
+          metrics[0].descriptor.name,
+          'db.client.operation.duration'
+        );
+        assert.strictEqual(
+          metrics[0].descriptor.description,
+          'Duration of database client operations.'
+        );
+        const dataPoint = metrics[0].dataPoints[0];
+        assert.strictEqual(
+          dataPoint.attributes[SEMATTRS_DB_SYSTEM],
+          DBSYSTEMVALUES_POSTGRESQL
+        );
+        assert.strictEqual(dataPoint.attributes[ATTR_ERROR_TYPE], undefined);
+
+        const v = (dataPoint as DataPoint<Histogram>).value;
+        v.min = v.min ? v.min : 0;
+        v.max = v.max ? v.max : 0;
+        assert.equal(
+          v.min > 0,
+          true,
+          'expect min value for Histogram to be greater than 0'
+        );
+        assert.equal(
+          v.max > 0,
+          true,
+          'expect max value for Histogram to be greater than 0'
+        );
+        done();
+      });
+    });
+
+    it('should generate db.client.operation.duration metric with error attribute', done => {
+      client.query('SELECT test()', async (err, ret) => {
+        assert.notEqual(err, null);
+        const { resourceMetrics, errors } = await metricReader.collect();
+        assert.deepEqual(
+          errors,
+          [],
+          'expected no errors from the callback during metric collection'
+        );
+
+        const metrics = resourceMetrics.scopeMetrics[0].metrics;
+        assert.strictEqual(
+          metrics[0].descriptor.name,
+          'db.client.operation.duration'
+        );
+        assert.strictEqual(
+          metrics[0].descriptor.description,
+          'Duration of database client operations.'
+        );
+        const dataPoint = metrics[0].dataPoints[0];
+        assert.strictEqual(
+          dataPoint.attributes[SEMATTRS_DB_SYSTEM],
+          DBSYSTEMVALUES_POSTGRESQL
+        );
+        assert.strictEqual(
+          dataPoint.attributes[ATTR_ERROR_TYPE],
+          'function test() does not exist'
+        );
+
+        const v = (dataPoint as DataPoint<Histogram>).value;
+        v.min = v.min ? v.min : 0;
+        v.max = v.max ? v.max : 0;
+        assert.equal(
+          v.min > 0,
+          true,
+          'expect min value for Histogram to be greater than 0'
+        );
+        assert.equal(
+          v.max > 0,
+          true,
+          'expect max value for Histogram to be greater than 0'
+        );
+        done();
+      });
+    });
+  });
 });