hanzokv/go
1
0
Fork 0
mirror of https://github.com/hanzokv/go.git synced 2026-09-24 00:31:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(otel): otel-native: Fix metric instrument types (#3743)

Browse Source 
* Change metric type from gauge to upAndDownCounter

* Fix pending requests metric

* Fix go version

* Bump grpc version

* Fix UpDownCounter metric accounting bugs in connection pool

Fix two bugs in the UpDownCounter metric instrumentation:

1. Cache miss path double-counts idle connections (High Severity):
   newConn() records +1 idle for every new connection, but the cache miss
   path in getConn() only recorded +1 used without compensating -1 idle.
   This caused the idle counter to grow unboundedly. Fixed by adding
   -1 idle compensation in the cache miss path and in the hook failure
   path for new connections.

2. Hook-rejected idle connection causes incorrect state accounting
   (Medium Severity): When a hook rejects an idle connection without
   error, putConnWithoutTurn calls putConn which records -1 used, +1 idle.
   But the connection was never transitioned to used state, causing the
   used counter to go negative. Fixed by recording the idle->used
   transition before calling putConnWithoutTurn so putConn's used->idle
   accounting is balanced.

* Fix UpDownCounter metric gaps

* fix: guard putConn metric emission against closed connections to prevent idle counter drift

* fix: Non-pooled connections get wrong metric state attribution

* fix: Pool Close() double-decrements in-use connection counters

* fix: Monotonic counter suppressed inside pool-closed guard

* fix: Shutdown race causes permanent UpDownCounter metric drift

* fix: Connection closed metric double-counted during pool shutdown

* fix:Race between putConn and Close causes permanent counter drift

* fix:Missing metric decrement when putIdleConn delivers to closed pool

---------

Co-authored-by: Nedyalko Dyakov <[email protected]>
This commit is contained in:
ofekshenawa
2026-04-21 14:12:25 +03:00
committed by GitHub
co-authored by Nedyalko Dyakov
parent d6b95bcdcf
commit 6cc94ebe5c
6 changed files with 422 additions and 306 deletions
Download Patch File Download Diff File Expand all files Collapse all files
extra/redisotel-native/metrics.go
+78 -72
View File
@@ -7,7 +7,6 @@ import (
"net"
"strconv"
"strings"
"sync"
"time"
"github.com/redis/go-redis/v9"
@@ -38,31 +37,19 @@ func addServerPortIfNonDefault(attrs []attribute.KeyValue, serverPort string) []
return attrs
}
// poolInfo stores information about a registered main connection pool
type poolInfo struct {
name string
pool redis.Pooler
}
// pubsubPoolInfo stores information about a registered PubSub pool
type pubsubPoolInfo struct {
name string
pool redis.PubSubPooler
}
// metricsRecorder implements the otel.Recorder interface
type metricsRecorder struct {
operationDuration metric.Float64Histogram
connectionCountGauge metric.Int64ObservableGauge
connectionCount metric.Int64UpDownCounter // OTel semconv: UpDownCounter for connection.count
connectionCreateTime metric.Float64Histogram
connectionRelaxedTimeout metric.Int64UpDownCounter
connectionHandoff metric.Int64Counter
clientErrors metric.Int64Counter
maintenanceNotifications metric.Int64Counter
connectionWaitTime metric.Float64Histogram
connectionClosed metric.Int64Counter
connectionPendingReqsGauge metric.Int64ObservableGauge
connectionWaitTime metric.Float64Histogram
connectionClosed metric.Int64Counter
connectionPendingReqs metric.Int64UpDownCounter // OTel semconv: UpDownCounter for pending_requests
pubsubMessages metric.Int64Counter
@@ -70,11 +57,6 @@ type metricsRecorder struct {
// Configuration
cfg *config
// Pool registry for tracking multiple pools
poolsMu sync.RWMutex
pools []poolInfo
pubsubPools []pubsubPoolInfo
}
// RecordOperationDuration records db.client.operation.duration metric
@@ -849,62 +831,86 @@ func (r *metricsRecorder) RecordStreamLag(
r.streamLag.Record(ctx, lag.Seconds(), metric.WithAttributes(attrs...))
}
// RegisterPool implements the OTelPoolRegistrar interface.
// The pools are used by async gauge callbacks to pull statistics.
func (r *metricsRecorder) RegisterPool(poolName string, pool redis.Pooler) {
r.poolsMu.Lock()
defer r.poolsMu.Unlock()
// RecordConnectionCount records a change in connection count (UpDownCounter).
// Per OTel semantic conventions, db.client.connection.count is an UpDownCounter.
// delta: +1 when connection added, -1 when connection removed
// state: connection state (e.g., "idle", "used")
// isPubSub: true if this is a PubSub connection
func (r *metricsRecorder) RecordConnectionCount(
ctx context.Context,
delta int,
cn redis.ConnInfo,
state string,
isPubSub bool,
) {
if r.connectionCount == nil {
return
}
// Add pool to registry
r.pools = append(r.pools, poolInfo{
name: poolName,
pool: pool,
})
}
// Build attributes
attrs := []attribute.KeyValue{
attribute.String(AttrDBSystemName, DBSystemRedis),
getLibraryVersionAttr(),
attribute.String(AttrDBClientConnectionState, state),
attribute.Bool(AttrRedisClientConnectionPubSub, isPubSub),
}
// UnregisterPool implements the OTelPoolRegistrar interface.
// This ensures async gauge callbacks don't try to access closed pools.
func (r *metricsRecorder) UnregisterPool(pool redis.Pooler) {
r.poolsMu.Lock()
defer r.poolsMu.Unlock()
// Use pool name from connection (set when connection was created)
if cn != nil {
poolName := cn.PoolName()
if poolName != "" {
attrs = append(attrs, attribute.String(AttrDBClientConnectionPoolName, poolName))
// Find and remove the pool from registry
for i, p := range r.pools {
if p.pool == pool {
// Remove by swapping with last element and truncating
r.pools[i] = r.pools[len(r.pools)-1]
r.pools = r.pools[:len(r.pools)-1]
return
// Extract server info from pool name
serverAddr, serverPort, _ := parsePoolName(poolName)
if serverAddr != "" {
attrs = append(attrs, attribute.String(AttrServerAddress, serverAddr))
}
attrs = addServerPortIfNonDefault(attrs, serverPort)
}
}
// Record the counter (delta can be +1 or -1)
r.connectionCount.Add(ctx, int64(delta), metric.WithAttributes(attrs...))
}
// RegisterPubSubPool implements the OTelPoolRegistrar interface.
// The pools are used by async gauge callbacks to pull statistics.
func (r *metricsRecorder) RegisterPubSubPool(poolName string, pool redis.PubSubPooler) {
r.poolsMu.Lock()
defer r.poolsMu.Unlock()
// Add PubSub pool to registry
r.pubsubPools = append(r.pubsubPools, pubsubPoolInfo{
name: poolName,
pool: pool,
})
}
// UnregisterPubSubPool implements the OTelPoolRegistrar interface.
// This ensures async gauge callbacks don't try to access closed pools.
func (r *metricsRecorder) UnregisterPubSubPool(pool redis.PubSubPooler) {
r.poolsMu.Lock()
defer r.poolsMu.Unlock()
// Find and remove the pool from registry
for i, p := range r.pubsubPools {
if p.pool == pool {
// Remove by swapping with last element and truncating
r.pubsubPools[i] = r.pubsubPools[len(r.pubsubPools)-1]
r.pubsubPools = r.pubsubPools[:len(r.pubsubPools)-1]
return
}
// RecordPendingRequests records a change in pending requests (UpDownCounter).
// Per OTel semantic conventions, db.client.connection.pending_requests is an UpDownCounter.
// delta: +1 when request starts waiting, -1 when request stops waiting
// poolName is passed explicitly because we may not have a connection yet when request starts
func (r *metricsRecorder) RecordPendingRequests(
ctx context.Context,
delta int,
cn redis.ConnInfo,
poolName string,
) {
if r.connectionPendingReqs == nil {
return
}
// Build attributes
attrs := []attribute.KeyValue{
attribute.String(AttrDBSystemName, DBSystemRedis),
getLibraryVersionAttr(),
}
// Use explicit pool name (preferred) or fall back to connection's pool name
effectivePoolName := poolName
if effectivePoolName == "" && cn != nil {
effectivePoolName = cn.PoolName()
}
if effectivePoolName != "" {
attrs = append(attrs, attribute.String(AttrDBClientConnectionPoolName, effectivePoolName))
// Extract server info from pool name
serverAddr, serverPort, _ := parsePoolName(effectivePoolName)
if serverAddr != "" {
attrs = append(attrs, attribute.String(AttrServerAddress, serverAddr))
}
attrs = addServerPortIfNonDefault(attrs, serverPort)
}
// Record the counter (delta can be +1 or -1)
r.connectionPendingReqs.Add(ctx, int64(delta), metric.WithAttributes(attrs...))
}
extra/redisotel-native/redisotel.go
+17 -172
View File
@@ -1,14 +1,12 @@
package redisotel
import (
"context"
"errors"
"fmt"
"strings"
"sync"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/metric"
"go.opentelemetry.io/otel/semconv/v1.38.0/dbconv"
@@ -198,13 +196,13 @@ func (o *ObservabilityInstance) createRecorder(meter metric.Meter, cfg config) (
operationDuration = operationDurationConv.Inst()
}
var connectionCountGauge metric.Int64ObservableGauge
var connectionCount metric.Int64UpDownCounter // OTel semconv: UpDownCounter
var connectionCreateTime metric.Float64Histogram
var connectionRelaxedTimeout metric.Int64UpDownCounter
var connectionHandoff metric.Int64Counter
if cfg.isMetricGroupEnabled(MetricGroupConnectionBasic) {
connectionCountGauge, err = meter.Int64ObservableGauge(
connectionCount, err = meter.Int64UpDownCounter(
dbconv.ClientConnectionCount{}.Name(),
metric.WithDescription(dbconv.ClientConnectionCount{}.Description()),
metric.WithUnit(dbconv.ClientConnectionCount{}.Unit()),
@@ -269,7 +267,7 @@ func (o *ObservabilityInstance) createRecorder(meter metric.Meter, cfg config) (
var connectionWaitTime metric.Float64Histogram
var connectionClosed metric.Int64Counter
var connectionPendingReqsGauge metric.Int64ObservableGauge
var connectionPendingReqs metric.Int64UpDownCounter // OTel semconv: UpDownCounter
if cfg.isMetricGroupEnabled(MetricGroupConnectionAdvanced) {
var connectionWaitTimeOpts []metric.Float64HistogramOption
@@ -294,7 +292,7 @@ func (o *ObservabilityInstance) createRecorder(meter metric.Meter, cfg config) (
return nil, fmt.Errorf("failed to create connection closed metric: %w", err)
}
connectionPendingReqsGauge, err = meter.Int64ObservableGauge(
connectionPendingReqs, err = meter.Int64UpDownCounter(
dbconv.ClientConnectionPendingRequests{}.Name(),
metric.WithDescription(dbconv.ClientConnectionPendingRequests{}.Description()),
metric.WithUnit(dbconv.ClientConnectionPendingRequests{}.Unit()),
@@ -341,26 +339,19 @@ func (o *ObservabilityInstance) createRecorder(meter metric.Meter, cfg config) (
// Create recorder
recorder := &metricsRecorder{
operationDuration: operationDuration,
connectionCountGauge: connectionCountGauge,
connectionCreateTime: connectionCreateTime,
connectionRelaxedTimeout: connectionRelaxedTimeout,
connectionHandoff: connectionHandoff,
clientErrors: clientErrors,
maintenanceNotifications: maintenanceNotifications,
connectionWaitTime: connectionWaitTime,
connectionClosed: connectionClosed,
connectionPendingReqsGauge: connectionPendingReqsGauge,
pubsubMessages: pubsubMessages,
streamLag: streamLag,
cfg: &cfg,
pools: make([]poolInfo, 0),
}
// Register async callbacks for ObservableGauges
// These callbacks will pull stats from registered pools
if err := o.registerAsyncCallbacks(meter, recorder); err != nil {
return nil, fmt.Errorf("failed to register async callbacks: %w", err)
operationDuration: operationDuration,
connectionCount: connectionCount,
connectionCreateTime: connectionCreateTime,
connectionRelaxedTimeout: connectionRelaxedTimeout,
connectionHandoff: connectionHandoff,
clientErrors: clientErrors,
maintenanceNotifications: maintenanceNotifications,
connectionWaitTime: connectionWaitTime,
connectionClosed: connectionClosed,
connectionPendingReqs: connectionPendingReqs,
pubsubMessages: pubsubMessages,
streamLag: streamLag,
cfg: &cfg,
}
return recorder, nil
@@ -384,149 +375,3 @@ func parsePoolName(poolName string) (string, string, string) {
host, port := parseAddr(addrPart)
return host, port, dbIndex
}
func (o *ObservabilityInstance) registerAsyncCallbacks(meter metric.Meter, recorder *metricsRecorder) error {
// Register connection count gauge callback
if recorder.connectionCountGauge != nil {
_, err := meter.RegisterCallback(
func(ctx context.Context, observer metric.Observer) error {
recorder.poolsMu.RLock()
pools := recorder.pools
pubsubPools := recorder.pubsubPools
recorder.poolsMu.RUnlock()
// Iterate over all registered main pools
for _, poolInfo := range pools {
stats := poolInfo.pool.PoolStats()
if stats == nil {
continue
}
// Extract server info from pool name
serverAddr, serverPort, _ := parsePoolName(poolInfo.name)
// Build base attributes
baseAttrs := []attribute.KeyValue{
attribute.String(AttrDBSystemName, DBSystemRedis),
getLibraryVersionAttr(),
}
if serverAddr != "" {
baseAttrs = append(baseAttrs, attribute.String(AttrServerAddress, serverAddr))
}
if serverPort != "" && serverPort != "6379" {
baseAttrs = append(baseAttrs, attribute.String(AttrServerPort, serverPort))
}
// Add pool name
baseAttrs = append(baseAttrs, attribute.String(AttrDBClientConnectionPoolName, poolInfo.name))
// Observe idle connections
idleAttrs := append([]attribute.KeyValue{}, baseAttrs...)
idleAttrs = append(idleAttrs,
attribute.String(AttrDBClientConnectionState, ConnectionStateIdle),
attribute.Bool(AttrRedisClientConnectionPubSub, false),
)
observer.ObserveInt64(recorder.connectionCountGauge, int64(stats.IdleConns),
metric.WithAttributes(idleAttrs...))
// Observe used connections
usedConns := stats.TotalConns - stats.IdleConns
usedAttrs := append([]attribute.KeyValue{}, baseAttrs...)
usedAttrs = append(usedAttrs,
attribute.String(AttrDBClientConnectionState, ConnectionStateUsed),
attribute.Bool(AttrRedisClientConnectionPubSub, false),
)
observer.ObserveInt64(recorder.connectionCountGauge, int64(usedConns),
metric.WithAttributes(usedAttrs...))
}
for _, pubsubPoolInfo := range pubsubPools {
stats := pubsubPoolInfo.pool.Stats()
if stats == nil {
continue
}
// Build base attributes
baseAttrs := []attribute.KeyValue{
attribute.String(AttrDBSystemName, DBSystemRedis),
getLibraryVersionAttr(),
attribute.String(AttrDBClientConnectionPoolName, pubsubPoolInfo.name),
}
// PubSub pools report Active connections (not idle/used split
// We'll report Active as "used" and 0 as "idle"
idleAttrs := append([]attribute.KeyValue{}, baseAttrs...)
idleAttrs = append(idleAttrs,
attribute.String(AttrDBClientConnectionState, ConnectionStateIdle),
attribute.Bool(AttrRedisClientConnectionPubSub, true),
)
observer.ObserveInt64(recorder.connectionCountGauge, 0,
metric.WithAttributes(idleAttrs...))
usedAttrs := append([]attribute.KeyValue{}, baseAttrs...)
usedAttrs = append(usedAttrs,
attribute.String(AttrDBClientConnectionState, ConnectionStateUsed),
attribute.Bool(AttrRedisClientConnectionPubSub, true),
)
observer.ObserveInt64(recorder.connectionCountGauge, int64(stats.Active),
metric.WithAttributes(usedAttrs...))
}
return nil
},
recorder.connectionCountGauge,
)
if err != nil {
return fmt.Errorf("failed to register connection count callback: %w", err)
}
}
// Register pending requests gauge callback
if recorder.connectionPendingReqsGauge != nil {
_, err := meter.RegisterCallback(
func(ctx context.Context, observer metric.Observer) error {
recorder.poolsMu.RLock()
pools := recorder.pools
recorder.poolsMu.RUnlock()
// Iterate over all registered pools
for _, poolInfo := range pools {
stats := poolInfo.pool.PoolStats()
if stats == nil {
continue
}
// Extract server info from pool name
serverAddr, serverPort, _ := parsePoolName(poolInfo.name)
// Build base attributes
baseAttrs := []attribute.KeyValue{
attribute.String(AttrDBSystemName, DBSystemRedis),
getLibraryVersionAttr(),
}
if serverAddr != "" {
baseAttrs = append(baseAttrs, attribute.String(AttrServerAddress, serverAddr))
}
if serverPort != "" && serverPort != "6379" {
baseAttrs = append(baseAttrs, attribute.String(AttrServerPort, serverPort))
}
// Add pool name
baseAttrs = append(baseAttrs, attribute.String(AttrDBClientConnectionPoolName, poolInfo.name))
// Observe pending requests count
observer.ObserveInt64(recorder.connectionPendingReqsGauge, int64(stats.PendingRequests),
metric.WithAttributes(baseAttrs...))
}
return nil
},
recorder.connectionPendingReqsGauge,
)
if err != nil {
return fmt.Errorf("failed to register pending requests callback: %w", err)
}
}
return nil
}
internal/otel/metrics.go
+19
View File
@@ -85,6 +85,17 @@ type Recorder interface {
// consumerGroup: name of the consumer group
// consumerName: name of the consumer
RecordStreamLag(ctx context.Context, lag time.Duration, cn *pool.Conn, streamName, consumerGroup, consumerName string)
// RecordConnectionCount records a change in connection count (UpDownCounter)
// delta: +1 when connection added, -1 when connection removed
// state: connection state (e.g., "idle", "used")
// isPubSub: true if this is a PubSub connection
RecordConnectionCount(ctx context.Context, delta int, cn *pool.Conn, state string, isPubSub bool)
// RecordPendingRequests records a change in pending requests (UpDownCounter)
// delta: +1 when request starts waiting, -1 when request stops waiting
// poolName is passed explicitly because we may not have a connection yet when request starts
RecordPendingRequests(ctx context.Context, delta int, cn *pool.Conn, poolName string)
}
type PubSubPooler interface {
@@ -193,6 +204,12 @@ func SetGlobalRecorder(r Recorder) {
ConnectionClosed: func(ctx context.Context, cn *pool.Conn, reason string, err error) {
getRecorder().RecordConnectionClosed(ctx, cn, reason, err)
},
ConnectionCount: func(ctx context.Context, delta int, cn *pool.Conn, state string, isPubSub bool) {
getRecorder().RecordConnectionCount(ctx, delta, cn, state, isPubSub)
},
PendingRequests: func(ctx context.Context, delta int, cn *pool.Conn, poolName string) {
getRecorder().RecordPendingRequests(ctx, delta, cn, poolName)
},
})
}
@@ -246,6 +263,8 @@ func (noopRecorder) RecordPubSubMessage(context.Context, *pool.Conn, string, str
func (noopRecorder) RecordStreamLag(context.Context, time.Duration, *pool.Conn, string, string, string) {
}
func (noopRecorder) RecordConnectionCount(context.Context, int, *pool.Conn, string, bool) {}
func (noopRecorder) RecordPendingRequests(context.Context, int, *pool.Conn, string) {}
// RegisterPools registers connection pools with the global recorder.
func RegisterPools(connPool pool.Pooler, pubSubPool PubSubPooler, addr string) {
internal/pool/pool.go
+252 -61
View File
@@ -101,6 +101,15 @@ var (
// Parameters: ctx, cn, reason, err
metricConnectionClosedCallback func(ctx context.Context, cn *Conn, reason string, err error)
// Global metric callback for connection count changes (UpDownCounter)
// Parameters: ctx, delta (+1/-1), cn, state, isPubSub
metricConnectionCountCallback func(ctx context.Context, delta int, cn *Conn, state string, isPubSub bool)
// Global metric callback for pending requests changes (UpDownCounter)
// Parameters: ctx, delta (+1/-1), cn, poolName
// poolName is passed explicitly because we may not have a connection yet when request starts
metricPendingRequestsCallback func(ctx context.Context, delta int, cn *Conn, poolName string)
// errPanicInDial is returned when a panic occurs in the dial function.
errPanicInQueuedNewConn = errors.New("panic in queuedNewConn")
@@ -146,6 +155,17 @@ type MetricCallbacks struct {
// ConnectionClosed is called when a connection is closed
ConnectionClosed func(ctx context.Context, cn *Conn, reason string, err error)
// ConnectionCount is called when connection count changes (UpDownCounter)
// delta: +1 when connection added, -1 when connection removed
// state: connection state (e.g., "idle", "used")
// isPubSub: true if this is a PubSub connection
ConnectionCount func(ctx context.Context, delta int, cn *Conn, state string, isPubSub bool)
// PendingRequests is called when pending requests count changes (UpDownCounter)
// delta: +1 when request starts waiting, -1 when request stops waiting
// poolName is passed explicitly because we may not have a connection yet when request starts
PendingRequests func(ctx context.Context, delta int, cn *Conn, poolName string)
}
// SetAllMetricCallbacks sets all metric callbacks atomically.
@@ -170,6 +190,8 @@ func SetAllMetricCallbacks(callbacks *MetricCallbacks) {
metricMaintenanceNotificationCallback = nil
metricConnectionWaitTimeCallback = nil
metricConnectionClosedCallback = nil
metricConnectionCountCallback = nil
metricPendingRequestsCallback = nil
return
}
@@ -180,6 +202,8 @@ func SetAllMetricCallbacks(callbacks *MetricCallbacks) {
metricMaintenanceNotificationCallback = callbacks.MaintenanceNotification
metricConnectionWaitTimeCallback = callbacks.ConnectionWaitTime
metricConnectionClosedCallback = callbacks.ConnectionClosed
metricConnectionCountCallback = callbacks.ConnectionCount
metricPendingRequestsCallback = callbacks.PendingRequests
}
// getMetricConnectionStateChangeCallback returns the metric callback for connection state changes.
@@ -255,6 +279,22 @@ func getMetricConnectionClosedCallback() func(ctx context.Context, cn *Conn, rea
return cb
}
// getMetricConnectionCountCallback returns the metric callback for connection count changes (UpDownCounter).
func getMetricConnectionCountCallback() func(ctx context.Context, delta int, cn *Conn, state string, isPubSub bool) {
metricCallbackMu.RLock()
cb := metricConnectionCountCallback
metricCallbackMu.RUnlock()
return cb
}
// getMetricPendingRequestsCallback returns the metric callback for pending requests changes (UpDownCounter).
func getMetricPendingRequestsCallback() func(ctx context.Context, delta int, cn *Conn, poolName string) {
metricCallbackMu.RLock()
cb := metricPendingRequestsCallback
metricCallbackMu.RUnlock()
return cb
}
// Stats contains pool state information and accumulated stats.
type Stats struct {
Hits uint32 // number of times free connection was found in the pool
@@ -513,6 +553,12 @@ func (p *ConnPool) addIdleConn() error {
p.conns[cn.GetID()] = cn
p.idleConns = append(p.idleConns, cn)
// Record connection count increment (new idle connection from min-idle prewarm)
if cb := getMetricConnectionCountCallback(); cb != nil {
cb(context.Background(), 1, cn, "idle", false)
}
return nil
}
@@ -577,9 +623,14 @@ func (p *ConnPool) newConn(ctx context.Context, pooled bool) (*Conn, error) {
}
}
// Notify metrics: new connection created and idle
// All new connections start as "used" metrically. For the miss path in getConn,
// this is the final state. For putIdleConn (undelivered conn), a used→idle
// transition is emitted when it's added to idleConns.
if cb := getMetricConnectionStateChangeCallback(); cb != nil {
cb(ctx, cn, "", MetricStateIdle)
cb(ctx, cn, "", MetricStateUsed)
}
if cb := getMetricConnectionCountCallback(); cb != nil {
cb(ctx, 1, cn, "used", false)
}
return cn, nil
@@ -758,12 +809,21 @@ func (p *ConnPool) getConn(ctx context.Context) (cn *Conn, err error) {
}
// Track pending requests in pool stats
// NOTE: We only track in stats, not via callback. The AsyncGauge reads stats directly.
atomic.AddUint32(&p.stats.PendingRequests, 1)
// Record pending request increment (UpDownCounter)
// Pass pool name explicitly since we don't have a connection yet
poolName := p.cfg.Name
if cb := getMetricPendingRequestsCallback(); cb != nil {
cb(ctx, 1, nil, poolName)
}
defer func() {
if err != nil {
// Failed to get connection, decrement pending requests
atomic.AddUint32(&p.stats.PendingRequests, ^uint32(0)) // -1
// Record pending request decrement on failure
if cb := getMetricPendingRequestsCallback(); cb != nil {
cb(ctx, -1, nil, poolName)
}
}
}()
@@ -801,6 +861,17 @@ func (p *ConnPool) getConn(ctx context.Context) (cn *Conn, err error) {
p.connsMu.Lock()
cn, err = p.popIdle()
if cn != nil {
// Emit idle→used transition inside the lock so Close() sees
// consistent state (conn removed from idleConns = "used").
if cb := getMetricConnectionStateChangeCallback(); cb != nil {
cb(ctx, cn, MetricStateIdle, MetricStateUsed)
}
if cb := getMetricConnectionCountCallback(); cb != nil {
cb(ctx, -1, cn, "idle", false)
cb(ctx, 1, cn, "used", false)
}
}
p.connsMu.Unlock()
if err != nil {
@@ -813,8 +884,8 @@ func (p *ConnPool) getConn(ctx context.Context) (cn *Conn, err error) {
}
if !p.isHealthyConn(cn, nowNs) {
// Connection was popped from idle pool, so fromState is MetricStateIdle
_ = p.CloseConn(ctx, cn, CloseReasonStale, MetricStateIdle)
// Connection was already transitioned to MetricStateUsed under the lock above.
_ = p.CloseConn(ctx, cn, CloseReasonStale, MetricStateUsed)
continue
}
@@ -825,12 +896,13 @@ func (p *ConnPool) getConn(ctx context.Context) (cn *Conn, err error) {
if hookErr != nil || !acceptConn {
if hookErr != nil {
internal.Logger.Printf(ctx, "redis: connection pool: failed to process idle connection by hook: %v", hookErr)
// Connection was popped from idle pool, so fromState is MetricStateIdle
_ = p.CloseConn(ctx, cn, CloseReasonHookError, MetricStateIdle)
// Connection was already transitioned to MetricStateUsed under the lock above.
_ = p.CloseConn(ctx, cn, CloseReasonHookError, MetricStateUsed)
} else {
internal.Logger.Printf(ctx, "redis: connection pool: conn[%d] rejected by hook, returning to pool", cn.GetID())
// Connection is already in MetricStateUsed (transitioned under the lock above).
// Return connection to pool without freeing the turn that this Get() call holds.
// We use putConnWithoutTurn() to run all the Put hooks and logic without freeing a turn.
// putConnWithoutTurn will emit used→idle transition.
p.putConnWithoutTurn(ctx, cn)
cn = nil
}
@@ -840,19 +912,17 @@ func (p *ConnPool) getConn(ctx context.Context) (cn *Conn, err error) {
atomic.AddUint32(&p.stats.Hits, 1)
// Notify metrics: connection moved from idle to used
if cb := getMetricConnectionStateChangeCallback(); cb != nil {
cb(ctx, cn, MetricStateIdle, MetricStateUsed)
}
// Record wait time (use cached callback from above)
if waitTimeCallback != nil {
waitTimeCallback(ctx, waitDuration, cn)
}
// Decrement pending requests (connection acquired successfully)
// NOTE: We only track in stats, not via callback. The AsyncGauge reads stats directly.
atomic.AddUint32(&p.stats.PendingRequests, ^uint32(0)) // -1
// Record pending request decrement (UpDownCounter)
if cb := getMetricPendingRequestsCallback(); cb != nil {
cb(ctx, -1, cn, poolName)
}
return cn, nil
}
@@ -873,18 +943,26 @@ func (p *ConnPool) getConn(ctx context.Context) (cn *Conn, err error) {
// both errors and accept=false mean a hook rejected the connection
// this should not happen with a new connection, but we handle it gracefully
if err != nil || !acceptConn {
// Failed to process connection, discard it
internal.Logger.Printf(ctx, "redis: connection pool: failed to process new connection conn[%d] by hook: accept=%v, err=%v", newcn.GetID(), acceptConn, err)
// New connection was recorded as "" → MetricStateIdle in newConn, so fromState is MetricStateIdle
_ = p.CloseConn(ctx, newcn, CloseReasonHookError, MetricStateIdle)
// newConn emitted +1 used; CloseConn will emit -1 used if we own the removal.
_ = p.CloseConn(ctx, newcn, CloseReasonHookError, MetricStateUsed)
return nil, err
}
// Record connection creation time metric when hooks are used.
// When hookManager is set, ProcessOnGet initializes the connection (AUTH/HELLO),
// causing IsInited()=true. This means _getConn() in redis.go will take the
// early return path and never reach its create time recording.
// When hookManager is nil, _getConn() handles both initialization and create time recording.
if dialStartNs := newcn.GetDialStartNs(); dialStartNs > 0 {
if cb := GetMetricConnectionCreateTimeCallback(); cb != nil {
duration := time.Duration(time.Now().UnixNano() - dialStartNs)
cb(ctx, duration, newcn)
}
}
}
// Notify metrics: new connection is created and used
if cb := getMetricConnectionStateChangeCallback(); cb != nil {
cb(ctx, newcn, "", MetricStateUsed)
}
// newConn already emitted +1 used, so no transition needed here.
// Record wait time (use cached callback from above)
if waitTimeCallback != nil {
@@ -892,8 +970,11 @@ func (p *ConnPool) getConn(ctx context.Context) (cn *Conn, err error) {
}
// Decrement pending requests (connection acquired successfully)
// NOTE: We only track in stats, not via callback. The AsyncGauge reads stats directly.
atomic.AddUint32(&p.stats.PendingRequests, ^uint32(0)) // -1
// Record pending request decrement (UpDownCounter)
if cb := getMetricPendingRequestsCallback(); cb != nil {
cb(ctx, -1, newcn, poolName)
}
return newcn, nil
}
@@ -992,14 +1073,24 @@ func (p *ConnPool) putIdleConn(ctx context.Context, cn *Conn) bool {
defer p.connsMu.Unlock()
if p.closed() {
_ = cn.Close()
// Don't close here — this connection is still in p.conns and Close()
// will handle closing it and emitting the correct metric decrements.
// We just skip adding it to idleConns.
return true
}
// poolSize is increased in newConn
p.idleConns = append(p.idleConns, cn)
p.idleConnsLen.Add(1)
// Connection was created as "used" in newConn; transition to idle.
if cb := getMetricConnectionStateChangeCallback(); cb != nil {
cb(ctx, cn, MetricStateUsed, MetricStateIdle)
}
if cb := getMetricConnectionCountCallback(); cb != nil {
cb(ctx, -1, cn, "used", false)
cb(ctx, 1, cn, "idle", false)
}
return true
}
@@ -1160,6 +1251,7 @@ func (p *ConnPool) putConn(ctx context.Context, cn *Conn, freeTurn bool) {
}
var shouldCloseConn bool
var removedFromPool bool
if p.cfg.MaxIdleConns == 0 || p.idleConnsLen.Load() < p.cfg.MaxIdleConns {
// Hot path optimization: try fast IN_USE → IDLE transition
@@ -1184,7 +1276,7 @@ func (p *ConnPool) putConn(ctx context.Context, cn *Conn, freeTurn bool) {
case StateClosed:
internal.Logger.Printf(ctx, "Unexpected conn[%d] state changed by hook to %v, closing it", cn.GetID(), currentState)
shouldCloseConn = true
p.removeConnWithLock(cn)
removedFromPool = p.removeConnWithLock(cn)
default:
// Pool as-is
internal.Logger.Printf(ctx, "Unexpected conn[%d] state changed by hook to %v, pooling as-is", cn.GetID(), currentState)
@@ -1195,34 +1287,73 @@ func (p *ConnPool) putConn(ctx context.Context, cn *Conn, freeTurn bool) {
// put them at the opposite end of the queue
// Optimization: if we just transitioned to IDLE, we know it's usable - skip the check
if !transitionedToIdle && !cn.IsUsable() {
if p.cfg.PoolFIFO {
p.connsMu.Lock()
p.idleConns = append(p.idleConns, cn)
p.connsMu.Lock()
// Check if Close() already removed this connection from p.conns.
// If so, skip the append and metrics — Close() already accounted for it.
if _, inPool := p.conns[cn.GetID()]; inPool {
if p.cfg.PoolFIFO {
p.idleConns = append(p.idleConns, cn)
} else {
p.idleConns = append([]*Conn{cn}, p.idleConns...)
}
if cb := getMetricConnectionStateChangeCallback(); cb != nil {
cb(ctx, cn, MetricStateUsed, MetricStateIdle)
}
if cb := getMetricConnectionCountCallback(); cb != nil {
cb(ctx, -1, cn, "used", false)
cb(ctx, 1, cn, "idle", false)
}
p.connsMu.Unlock()
p.idleConnsLen.Add(1)
} else {
p.connsMu.Lock()
p.idleConns = append([]*Conn{cn}, p.idleConns...)
shouldCloseConn = true
p.connsMu.Unlock()
}
p.idleConnsLen.Add(1)
} else if !shouldCloseConn {
p.connsMu.Lock()
p.idleConns = append(p.idleConns, cn)
p.connsMu.Unlock()
p.idleConnsLen.Add(1)
if _, inPool := p.conns[cn.GetID()]; inPool {
p.idleConns = append(p.idleConns, cn)
if cb := getMetricConnectionStateChangeCallback(); cb != nil {
cb(ctx, cn, MetricStateUsed, MetricStateIdle)
}
if cb := getMetricConnectionCountCallback(); cb != nil {
cb(ctx, -1, cn, "used", false)
cb(ctx, 1, cn, "idle", false)
}
p.connsMu.Unlock()
p.idleConnsLen.Add(1)
} else {
shouldCloseConn = true
p.connsMu.Unlock()
}
}
// Notify metrics: connection moved from used to idle
if cb := getMetricConnectionStateChangeCallback(); cb != nil {
cb(ctx, cn, MetricStateUsed, MetricStateIdle)
if shouldCloseConn {
// Connection was removed (e.g., hook set state to StateClosed).
// Only emit if we actually removed it from the map (not already taken by Close()).
if removedFromPool {
if cb := getMetricConnectionStateChangeCallback(); cb != nil {
cb(ctx, cn, MetricStateUsed, "")
}
if cb := getMetricConnectionCountCallback(); cb != nil {
cb(ctx, -1, cn, "used", false)
}
}
}
} else {
shouldCloseConn = true
p.removeConnWithLock(cn)
removedFromPool = p.removeConnWithLock(cn)
// Notify metrics: connection removed (used -> nothing)
if cb := getMetricConnectionStateChangeCallback(); cb != nil {
cb(ctx, cn, MetricStateUsed, "")
// Only emit if we actually removed it from the map (not already taken by Close()).
if removedFromPool {
// Notify metrics: connection removed (used -> nothing)
if cb := getMetricConnectionStateChangeCallback(); cb != nil {
cb(ctx, cn, MetricStateUsed, "")
}
// Record connection count decrement (connection removed while in used state)
if cb := getMetricConnectionCountCallback(); cb != nil {
cb(ctx, -1, cn, "used", false)
}
}
}
@@ -1231,6 +1362,13 @@ func (p *ConnPool) putConn(ctx context.Context, cn *Conn, freeTurn bool) {
}
if shouldCloseConn {
// Only emit connection closed if we actually owned the removal.
// If removedFromPool is false, Close() already emitted connectionClosed for this conn.
if removedFromPool {
if cb := getMetricConnectionClosedCallback(); cb != nil {
cb(ctx, cn, "conn_pool_close", nil)
}
}
_ = p.closeConn(cn)
}
@@ -1258,24 +1396,35 @@ func (p *ConnPool) removeConnInternal(ctx context.Context, cn *Conn, reason erro
hookManager.ProcessOnRemove(ctx, cn, reason)
}
p.removeConnWithLock(cn)
removed := p.removeConnWithLock(cn)
if freeTurn {
p.freeTurn()
}
// Notify metrics: connection removed (assume from used state)
if cb := getMetricConnectionStateChangeCallback(); cb != nil {
cb(ctx, cn, MetricStateUsed, "")
// Only emit metric decrements if we actually removed the connection from the map.
// If removed is false, Close() already removed it and emitted the -1 delta.
if removed {
// Notify metrics: connection removed (assume from used state)
if cb := getMetricConnectionStateChangeCallback(); cb != nil {
cb(ctx, cn, MetricStateUsed, "")
}
// Record connection count decrement (connection removed, assume from used state)
if cb := getMetricConnectionCountCallback(); cb != nil {
cb(ctx, -1, cn, "used", false)
}
}
// Record connection closed
if cb := getMetricConnectionClosedCallback(); cb != nil {
reasonStr := "unknown"
if reason != nil {
reasonStr = reason.Error()
// Only emit connection closed if we actually owned the removal.
// If removed is false, Close() already emitted connectionClosed for this conn.
if removed {
if cb := getMetricConnectionClosedCallback(); cb != nil {
reasonStr := "unknown"
if reason != nil {
reasonStr = reason.Error()
}
cb(ctx, cn, reasonStr, reason)
}
cb(ctx, cn, reasonStr, reason)
}
_ = p.closeConn(cn)
@@ -1291,29 +1440,51 @@ func (p *ConnPool) removeConnInternal(ctx context.Context, cn *Conn, reason erro
// - reason: why the connection is being closed (use CloseReason* constants)
// - fromState: the metric state the connection was in (use MetricState* constants)
func (p *ConnPool) CloseConn(ctx context.Context, cn *Conn, reason string, fromState string) error {
p.removeConnWithLock(cn)
removed := p.removeConnWithLock(cn)
// Record connection state change: connection is being removed from the specified state
if cb := getMetricConnectionStateChangeCallback(); cb != nil && fromState != "" {
cb(ctx, cn, fromState, "")
// Only emit UpDownCounter decrements if we actually removed the connection.
// If removed is false, Close() already removed it and emitted the -1 delta.
if removed {
// Record connection state change: connection is being removed from the specified state
if cb := getMetricConnectionStateChangeCallback(); cb != nil && fromState != "" {
cb(ctx, cn, fromState, "")
}
// Record connection count decrement (UpDownCounter) for the state the connection was in
if cb := getMetricConnectionCountCallback(); cb != nil && fromState != "" {
cb(ctx, -1, cn, fromState, false)
}
}
// Record connection closed metric with the specified reason
if cb := getMetricConnectionClosedCallback(); cb != nil {
cb(ctx, cn, reason, nil)
// Only emit connection closed if we actually owned the removal.
// If removed is false, Close() already emitted connectionClosed for this conn.
if removed {
if cb := getMetricConnectionClosedCallback(); cb != nil {
cb(ctx, cn, reason, nil)
}
}
return p.closeConn(cn)
}
func (p *ConnPool) removeConnWithLock(cn *Conn) {
// removeConnWithLock removes a connection from the pool under the connsMu lock.
// Returns true if the connection was actually present in p.conns and was removed,
// false if it was already gone (e.g., removed by Close()). Callers must use the
// return value to decide whether to emit metric decrements — this eliminates the
// shutdown race between Close() and concurrent removal paths.
func (p *ConnPool) removeConnWithLock(cn *Conn) bool {
p.connsMu.Lock()
defer p.connsMu.Unlock()
p.removeConn(cn)
return p.removeConn(cn)
}
func (p *ConnPool) removeConn(cn *Conn) {
// removeConn removes a connection from the pool's internal data structures.
// Returns true if the connection was present and removed, false otherwise.
func (p *ConnPool) removeConn(cn *Conn) bool {
cid := cn.GetID()
if _, exists := p.conns[cid]; !exists {
return false
}
delete(p.conns, cid)
atomic.AddUint32(&p.stats.StaleConns, 1)
@@ -1329,6 +1500,7 @@ func (p *ConnPool) removeConn(cn *Conn) {
}
}
}
return true
}
func (p *ConnPool) closeConn(cn *Conn) error {
@@ -1401,7 +1573,26 @@ func (p *ConnPool) Close() error {
var firstErr error
p.connsMu.Lock()
// Emit -1 for each connection. Since all idle↔used transitions happen
// under connsMu, the idleConns slice is the source of truth for state.
cb := getMetricConnectionCountCallback()
idleSet := make(map[uint64]struct{}, len(p.idleConns))
for _, cn := range p.idleConns {
idleSet[cn.GetID()] = struct{}{}
}
ctx := context.Background()
for _, cn := range p.conns {
if cb != nil {
if _, isIdle := idleSet[cn.GetID()]; isIdle {
cb(ctx, -1, cn, "idle", false)
} else {
cb(ctx, -1, cn, "used", false)
}
}
if closedCb := getMetricConnectionClosedCallback(); closedCb != nil {
closedCb(ctx, cn, "pool_shutdown", nil)
}
if err := p.closeConn(cn); err != nil && firstErr == nil {
firstErr = err
}
internal/pool/pubsub.go
+25 -1
View File
@@ -53,18 +53,42 @@ func (p *PubSubPool) NewConn(ctx context.Context, network string, addr string, c
func (p *PubSubPool) TrackConn(cn *Conn) {
atomic.AddUint32(&p.stats.Active, 1)
p.activeConns.Store(cn.GetID(), cn)
// Emit +1 used for PubSub connection
if cb := getMetricConnectionCountCallback(); cb != nil {
cb(context.Background(), 1, cn, "used", true)
}
}
func (p *PubSubPool) UntrackConn(cn *Conn) {
// LoadAndDelete ensures each connection is only decremented once,
// guarding against double-decrement if Close() already untracked it.
if _, loaded := p.activeConns.LoadAndDelete(cn.GetID()); !loaded {
return
}
atomic.AddUint32(&p.stats.Active, ^uint32(0))
atomic.AddUint32(&p.stats.Untracked, 1)
p.activeConns.Delete(cn.GetID())
// Emit -1 used for PubSub connection
if cb := getMetricConnectionCountCallback(); cb != nil {
cb(context.Background(), -1, cn, "used", true)
}
}
func (p *PubSubPool) Close() error {
p.closed.Store(true)
cb := getMetricConnectionCountCallback()
p.activeConns.Range(func(key, value interface{}) bool {
cn := value.(*Conn)
// Use LoadAndDelete to atomically claim ownership of this entry.
// If a concurrent UntrackConn already removed it, skip to avoid double-decrement.
if _, loaded := p.activeConns.LoadAndDelete(key); !loaded {
return true
}
atomic.AddUint32(&p.stats.Active, ^uint32(0))
atomic.AddUint32(&p.stats.Untracked, 1)
// Emit -1 used for each PubSub connection being closed
if cb != nil {
cb(context.Background(), -1, cn, "used", true)
}
_ = cn.Close()
return true
})
otel.go
+31
View File
@@ -79,6 +79,25 @@ type OTelRecorder interface {
RecordStreamLag(ctx context.Context, lag time.Duration, cn ConnInfo, streamName, consumerGroup, consumerName string)
}
// OTelConnectionCounter is an optional capability interface for recording
// connection count and pending request changes via UpDownCounters.
// Implementations of OTelRecorder can optionally implement this interface
// to receive connection count and pending request delta notifications.
// This is kept separate from OTelRecorder to avoid breaking existing
// third-party implementations when new methods are added.
type OTelConnectionCounter interface {
// RecordConnectionCount records a change in connection count (UpDownCounter)
// delta: +1 when connection added, -1 when connection removed
// state: connection state (e.g., "idle", "used")
// isPubSub: true if this is a PubSub connection
RecordConnectionCount(ctx context.Context, delta int, cn ConnInfo, state string, isPubSub bool)
// RecordPendingRequests records a change in pending requests (UpDownCounter)
// delta: +1 when request starts waiting, -1 when request stops waiting
// poolName is passed explicitly because we may not have a connection yet when request starts
RecordPendingRequests(ctx context.Context, delta int, cn ConnInfo, poolName string)
}
// This is used for async gauge metrics that need to pull stats from pools periodically.
type OTelPoolRegistrar interface {
// RegisterPool is called when a new client is created with its main connection pool.
@@ -163,6 +182,18 @@ func (a *otelRecorderAdapter) RecordStreamLag(ctx context.Context, lag time.Dura
a.recorder.RecordStreamLag(ctx, lag, toConnInfo(cn), streamName, consumerGroup, consumerName)
}
func (a *otelRecorderAdapter) RecordConnectionCount(ctx context.Context, delta int, cn *pool.Conn, state string, isPubSub bool) {
if counter, ok := a.recorder.(OTelConnectionCounter); ok {
counter.RecordConnectionCount(ctx, delta, toConnInfo(cn), state, isPubSub)
}
}
func (a *otelRecorderAdapter) RecordPendingRequests(ctx context.Context, delta int, cn *pool.Conn, poolName string) {
if counter, ok := a.recorder.(OTelConnectionCounter); ok {
counter.RecordPendingRequests(ctx, delta, toConnInfo(cn), poolName)
}
}
func (a *otelRecorderAdapter) RegisterPool(poolName string, p pool.Pooler) {
if registrar, ok := a.recorder.(OTelPoolRegistrar); ok {
registrar.RegisterPool(poolName, &poolerAdapter{p})