GODRIVER-2423 Include pinned connections in WaitQueueTimeoutError only for load balanced clusters. (#1353)

Author: qingyang-hu

x/mongo/driver/topology/errors.go

@@ -9,6 +9,7 @@ package topology
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"go.mongodb.org/mongo-driver/mongo/description"
 )
@@ -69,11 +70,17 @@ func (e ServerSelectionError) Unwrap() error {
 
 // WaitQueueTimeoutError represents a timeout when requesting a connection from the pool
 type WaitQueueTimeoutError struct {
-	Wrapped                      error
-	PinnedCursorConnections      uint64
-	PinnedTransactionConnections uint64
-	maxPoolSize                  uint64
-	totalConnectionCount         int
+	Wrapped              error
+	pinnedConnections    *pinnedConnections
+	maxPoolSize          uint64
+	totalConnections     int
+	availableConnections int
+	waitDuration         time.Duration
+}
+
+type pinnedConnections struct {
+	cursorConnections      uint64
+	transactionConnections uint64
 }
 
 // Error implements the error interface.
@@ -95,14 +102,19 @@ func (w WaitQueueTimeoutError) Error() string {
 		)
 	}
 
-	return fmt.Sprintf(
-		"%s; maxPoolSize: %d, connections in use by cursors: %d"+
-			", connections in use by transactions: %d, connections in use by other operations: %d",
-		errorMsg,
-		w.maxPoolSize,
-		w.PinnedCursorConnections,
-		w.PinnedTransactionConnections,
-		uint64(w.totalConnectionCount)-w.PinnedCursorConnections-w.PinnedTransactionConnections)
+	msg := fmt.Sprintf("%s; total connections: %d, maxPoolSize: %d, ", errorMsg, w.totalConnections, w.maxPoolSize)
+	if pinnedConnections := w.pinnedConnections; pinnedConnections != nil {
+		openConnectionCount := uint64(w.totalConnections) -
+			pinnedConnections.cursorConnections -
+			pinnedConnections.transactionConnections
+		msg += fmt.Sprintf("connections in use by cursors: %d, connections in use by transactions: %d, connections in use by other operations: %d, ",
+			pinnedConnections.cursorConnections,
+			pinnedConnections.transactionConnections,
+			openConnectionCount,
+		)
+	}
+	msg += fmt.Sprintf("idle connections: %d, wait duration: %s", w.availableConnections, w.waitDuration.String())
+	return msg
 }
 
 // Unwrap returns the underlying error.

x/mongo/driver/topology/pool.go

@@ -74,6 +74,7 @@ type poolConfig struct {
 	MaxConnecting    uint64
 	MaxIdleTime      time.Duration
 	MaintainInterval time.Duration
+	LoadBalanced     bool
 	PoolMonitor      *event.PoolMonitor
 	Logger           *logger.Logger
 	handshakeErrFn   func(error, uint64, *primitive.ObjectID)
@@ -93,6 +94,7 @@ type pool struct {
 	minSize       uint64
 	maxSize       uint64
 	maxConnecting uint64
+	loadBalanced  bool
 	monitor       *event.PoolMonitor
 	logger        *logger.Logger
 
@@ -206,6 +208,7 @@ func newPool(config poolConfig, connOpts ...ConnectionOption) *pool {
 		minSize:               config.MinPoolSize,
 		maxSize:               config.MaxPoolSize,
 		maxConnecting:         maxConnecting,
+		loadBalanced:          config.LoadBalanced,
 		monitor:               config.PoolMonitor,
 		logger:                config.Logger,
 		handshakeErrFn:        config.handshakeErrFn,
@@ -572,6 +575,7 @@ func (p *pool) checkOut(ctx context.Context) (conn *connection, err error) {
 	p.stateMu.RUnlock()
 
 	// Wait for either the wantConn to be ready or for the Context to time out.
+	start := time.Now()
 	select {
 	case <-w.ready:
 		if w.err != nil {
@@ -612,6 +616,8 @@ func (p *pool) checkOut(ctx context.Context) (conn *connection, err error) {
 		}
 		return w.conn, nil
 	case <-ctx.Done():
+		duration := time.Since(start)
+
 		if mustLogPoolMessage(p) {
 			keysAndValues := logger.KeyValues{
 				logger.KeyReason, logger.ReasonConnCheckoutFailedTimout,
@@ -628,13 +634,20 @@ func (p *pool) checkOut(ctx context.Context) (conn *connection, err error) {
 			})
 		}
 
-		return nil, WaitQueueTimeoutError{
-			Wrapped:                      ctx.Err(),
-			PinnedCursorConnections:      atomic.LoadUint64(&p.pinnedCursorConnections),
-			PinnedTransactionConnections: atomic.LoadUint64(&p.pinnedTransactionConnections),
-			maxPoolSize:                  p.maxSize,
-			totalConnectionCount:         p.totalConnectionCount(),
+		err := WaitQueueTimeoutError{
+			Wrapped:              ctx.Err(),
+			maxPoolSize:          p.maxSize,
+			totalConnections:     p.totalConnectionCount(),
+			availableConnections: p.availableConnectionCount(),
+			waitDuration:         duration,
+		}
+		if p.loadBalanced {
+			err.pinnedConnections = &pinnedConnections{
+				cursorConnections:      atomic.LoadUint64(&p.pinnedCursorConnections),
+				transactionConnections: atomic.LoadUint64(&p.pinnedTransactionConnections),
+			}
 		}
+		return nil, err
 	}
 }
 

x/mongo/driver/topology/server.go

@@ -176,6 +176,7 @@ func NewServer(addr address.Address, topologyID primitive.ObjectID, opts ...Serv
 		MaxConnecting:    cfg.maxConnecting,
 		MaxIdleTime:      cfg.poolMaxIdleTime,
 		MaintainInterval: cfg.poolMaintainInterval,
+		LoadBalanced:     cfg.loadBalanced,
 		PoolMonitor:      cfg.poolMonitor,
 		Logger:           cfg.logger,
 		handshakeErrFn:   s.ProcessHandshakeError,