add optional metrics args

Author: chrisdickinson

README.md

@@ -77,7 +77,52 @@ it if the promise is rejected. Atomics may be nested.
 
 Install a database `Session` on the domain `d`.
 
-#### `ConnPairFn := Function → Promise({connection, release})`
+##### `Options`
+
+Sessions accept the following options:
+
+* `maxConcurrency`: An integer specifying the maximum number of connections a
+  given session will make at a time. `0` is treated as `Infinity`. Defaults to
+  `Infinity`. *Note:* this number is implicitly bound by the size of the `pg`
+  connection pool. For example, even if the limit is set at `200`, if `pg`'s
+  pool size is limited to `10`, the upper limit will effectively be `10`.
+* `onSessionIdle()`: A function that is called whenever all requests for
+  connections have been satisfied. Note that this may happen while connections
+  are still open.
+* `onConnectionRequest(baton)`: A function accepting a baton object that is
+  called when a request for a connection is made.
+* `onConnectionStart(baton)`: A function acccepting a baton object that is
+  called when a request for a connection is fulfilled. The baton will be the
+  same object that was passed to a previous call to `onConnectionRequest`,
+  suitable for associating timing information.
+* `onConnectionFinish(baton, err)`: A function accepting a baton object and
+  an optional `err` parameter that will be called when a connection is released
+  back to the session.
+* `onTransactionRequest(baton, operation, args)`: A function accepting a baton,
+  function, and array of arguments representing the request for a transaction
+  session. Called coincident with `onConnectionRequest`.
+* `onTransactionStart(baton, operation, args)`: A function accepting a baton,
+  function, and array of arguments representing the fulfillment of a request
+  for a transaction session. Called before `BEGIN`, coincident with
+  `onConnectionStart`.
+* `onTransactionFinish(baton, operation, args, PromiseInspection)`:
+  A function accepting a baton, function, array of arguments, and a
+  [`PromiseInspection`][bluebird-inspection] representing the state of the
+  transaction. Called coincident with `onConnectionFinish`.
+* `onAtomicRequest(baton, operation, args)`: A function accepting a baton,
+  function, and array of arguments representing the request for an atomic
+  session.
+* `onAtomicStart(baton, operation, args)`: A function accepting a baton,
+  function, and array of arguments representing the fulfillment of a request
+  for an atomic session.
+* `onAtomicFinish(baton, operation, args, PromiseInspection)`:
+  A function accepting a baton, function, array of arguments, and a
+  [`PromiseInspection`][bluebird-inspection] representing the state of the
+  atomic transaction.
+
+All functions will default to `noop` if not provided.
+
+##### `ConnPairFn := Function → Promise({connection, release})`
 
 A function that returns a `Promise` for an object with `connection` and `release`
 properties, corresponding to the `client` and `done` parameters handed back by

db-session.js

@@ -14,12 +14,27 @@ class NoSessionAvailable extends Error {
   }
 }
 
+function noop () {
+}
+
 const api = module.exports = {
   install (domain, getConnection, opts) {
-    opts = opts || {}
+    opts = Object.assign({
+      maxConcurrency: Infinity,
+      onSessionIdle: noop,
+      onConnectionRequest: noop,
+      onConnectionStart: noop,
+      onConnectionFinish: noop,
+      onTransactionRequest: noop,
+      onTransactionStart: noop,
+      onTransactionFinish: noop,
+      onAtomicRequest: noop,
+      onAtomicStart: noop,
+      onAtomicFinish: noop
+    }, opts || {})
     DOMAIN_TO_SESSION.set(domain, new Session(
       getConnection,
-      opts.maxConcurrency
+      opts
     ))
   },
 
@@ -63,15 +78,29 @@ const api = module.exports = {
 // 3. atomic — grouped set of operations — parent transaction treats all connections performed
 //    as a single operation
 class Session {
-  constructor (getConnection, maxConcurrency) {
+  constructor (getConnection, opts) {
     this._getConnection = getConnection
-    this._activeConnections = 0
-    this._maxConcurrency = maxConcurrency || Infinity
+    this.activeConnections = 0
+    this.maxConcurrency = opts.maxConcurrency || Infinity
+    this.metrics = {
+      onSessionIdle: opts.onSessionIdle,
+      onConnectionRequest: opts.onConnectionRequest,
+      onConnectionStart: opts.onConnectionStart,
+      onConnectionFinish: opts.onConnectionFinish,
+      onTransactionRequest: opts.onTransactionRequest,
+      onTransactionStart: opts.onTransactionStart,
+      onTransactionFinish: opts.onTransactionFinish,
+      onAtomicRequest: opts.onAtomicRequest,
+      onAtomicStart: opts.onAtomicStart,
+      onAtomicFinish: opts.onAtomicFinish
+    }
     this.pending = []
   }
 
   getConnection () {
-    if (this._activeConnections === this._maxConcurrency) {
+    const baton = {}
+    this.metrics.onConnectionRequest(baton)
+    if (this.activeConnections === this.maxConcurrency) {
       // not using Promise.defer() here in case it gets deprecated by
       // bluebird.
       const pending = _defer()
@@ -80,24 +109,29 @@ class Session {
     }
 
     const connPair = Promise.resolve(this._getConnection())
-    ++this._activeConnections
+    ++this.activeConnections
 
-    return connPair.then(
-      pair => new SessionConnectionPair(pair, this)
-    )
+    return connPair.then(pair => {
+      this.metrics.onConnectionStart(baton)
+      return new SessionConnectionPair(pair, this, baton)
+    })
   }
 
   transaction (operation, args) {
+    const baton = {}
     const getConnPair = this.getConnection()
-    const getResult = Session$RunWrapped(connPair => {
-      return new TransactionSession(connPair)
+    this.metrics.onTransactionRequest(baton, operation, args)
+    const getResult = Session$RunWrapped(this, connPair => {
+      this.metrics.onTransactionStart(baton, operation, args)
+      return new TransactionSession(connPair, this.metrics)
     }, getConnPair, `BEGIN`, {
       success: `COMMIT`,
       failure: `ROLLBACK`
     }, operation, args)
 
     const releasePair = getConnPair.then(pair => {
       return getResult.reflect().then(result => {
+        this.metrics.onTransactionFinish(baton, operation, args, result)
         return result.isFulfilled()
           ? pair.release()
           : pair.release(result.reason())
@@ -114,16 +148,17 @@ class Session {
   }
 
   releasePair (pair, err) {
-    --this._activeConnections
+    --this.activeConnections
     pair.release(err)
   }
 }
 
 class TransactionSession {
-  constructor (connPair) {
+  constructor (connPair, metrics) {
     this.connectionPair = connPair
     this.inactive = false
     this.operation = Promise.resolve(true)
+    this.metrics = metrics
   }
 
   getConnection () {
@@ -148,17 +183,21 @@ class TransactionSession {
   }
 
   atomic (operation, args) {
+    const baton = {}
     const atomicConnPair = this.getConnection()
     const savepointName = getSavepointName(operation)
-    const getResult = Session$RunWrapped(connPair => {
-      return new AtomicSession(connPair, savepointName)
+    this.metrics.onAtomicRequest(baton, operation, args)
+    const getResult = Session$RunWrapped(this, connPair => {
+      this.metrics.onAtomicStart(baton, operation, args)
+      return new AtomicSession(connPair, this.metrics, savepointName)
     }, atomicConnPair, `SAVEPOINT ${savepointName}`, {
       success: `RELEASE SAVEPOINT ${savepointName}`,
       failure: `ROLLBACK TO SAVEPOINT ${savepointName}`
     }, operation, args)
 
     const releasePair = atomicConnPair.then(pair => {
       return getResult.reflect().then(result => {
+        this.metrics.onAtomicFinish(baton, operation, args, result)
         return result.isFulfilled()
           ? pair.release()
           : pair.release(result.reason())
@@ -170,13 +209,14 @@ class TransactionSession {
 }
 
 class AtomicSession extends TransactionSession {
-  constructor (connection, name) {
-    super(connection)
+  constructor (connection, metrics, name) {
+    super(connection, metrics)
     this.name = name
   }
 }
 
-function Session$RunWrapped (createSession,
+function Session$RunWrapped (parent,
+                             createSession,
                              getConnPair,
                              before,
                              after,

lib/session-connpair.js

@@ -1,9 +1,10 @@
 'use strict'
 
 module.exports = class SessionConnectionPair {
-  constructor (connPair, session) {
+  constructor (connPair, session, baton) {
     this.pair = connPair
     this.session = session
+    this.baton = baton
 
     // tightly bind "release", because we don't know
     // who will be calling it.
@@ -20,13 +21,15 @@ module.exports = class SessionConnectionPair {
 // the connection entirely. This lets us limit the concurrency
 // per-request, instead of globally.
 function release (conn, err) {
+  conn.session.metrics.onConnectionFinish(conn.baton, err)
   if (err) {
     return handleError(conn, err)
   }
   const next = conn.session.pending.shift()
   if (next) {
     return next.resolve(conn)
   }
+  conn.session.metrics.onSessionIdle()
   conn.session.releasePair(conn.pair, null)
 }