go/extra/redisotel-native/metrics.go

package redisotel

import (
	"context"
	"errors"
	"fmt"
	"net"
	"strconv"
	"strings"
	"time"

	"github.com/redis/go-redis/v9"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/metric"
)

// timeoutError is an interface for errors that have a Timeout() method
type timeoutError interface {
	Timeout() bool
}

const (
	// Library name for redis.client.library attribute
	libraryName = "go-redis"
)

// getLibraryVersionAttr returns the redis.client.library attribute
func getLibraryVersionAttr() attribute.KeyValue {
	return attribute.String(AttrRedisClientLibrary, fmt.Sprintf("%s:%s", libraryName, redis.Version()))
}

// addServerPortIfNonDefault adds server.port attribute if port is not the default (6379)
func addServerPortIfNonDefault(attrs []attribute.KeyValue, serverPort string) []attribute.KeyValue {
	if serverPort != "" && serverPort != "6379" {
		return append(attrs, attribute.String(AttrServerPort, serverPort))
	}
	return attrs
}

// metricsRecorder implements the otel.Recorder interface
type metricsRecorder struct {
	operationDuration        metric.Float64Histogram
	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
	connectionPendingReqs metric.Int64UpDownCounter // OTel semconv: UpDownCounter for pending_requests

	pubsubMessages metric.Int64Counter

	streamLag metric.Float64Histogram

	// Configuration
	cfg *config
}

// RecordOperationDuration records db.client.operation.duration metric
func (r *metricsRecorder) RecordOperationDuration(
	ctx context.Context,
	duration time.Duration,
	cmd redis.Cmder,
	attempts int,
	err error,
	cn redis.ConnInfo,
	dbIndex int,
) {
	if r.operationDuration == nil {
		return
	}

	// Check if command should be included
	if r.cfg != nil && !r.cfg.isCommandIncluded(cmd.Name()) {
		return
	}

	// Convert duration to seconds (OTel convention for duration metrics)
	durationSeconds := duration.Seconds()

	serverAddr, serverPort := extractServerInfo(cn)

	// Build attributes
	attrs := []attribute.KeyValue{
		// Required attributes
		attribute.String(AttrDBOperationName, cmd.FullName()),
		getLibraryVersionAttr(),
		attribute.Int(AttrRedisClientOperationRetryAttempts, attempts-1), // attempts-1 = retry count

		// Recommended attributes
		attribute.String(AttrDBSystemName, DBSystemRedis),
		attribute.String(AttrServerAddress, serverAddr),
		attribute.String(AttrDBNamespace, strconv.Itoa(dbIndex)),
	}

	// Add server.port if not default
	attrs = addServerPortIfNonDefault(attrs, serverPort)

	// Add network.peer.address and network.peer.port from connection
	if cn != nil {
		remoteAddr := cn.RemoteAddr()
		if remoteAddr != nil {
			peerAddr, peerPort := splitHostPort(remoteAddr.String())
			if peerAddr != "" {
				attrs = append(attrs, attribute.String(AttrNetworkPeerAddress, peerAddr))
			}
			if peerPort != "" {
				attrs = append(attrs, attribute.String(AttrNetworkPeerPort, peerPort))
			}
		}
	}

	if err != nil {
		attrs = append(attrs, attribute.String(AttrErrorType, classifyError(err)))
		attrs = append(attrs, attribute.String(AttrRedisClientErrorsCategory, getErrorCategory(err)))
		if statusCode := extractRedisErrorPrefix(err); statusCode != "" {
			attrs = append(attrs, attribute.String(AttrDBResponseStatusCode, statusCode))
		}
	}

	// Record the histogram
	r.operationDuration.Record(ctx, durationSeconds, metric.WithAttributes(attrs...))
}

// RecordPipelineOperationDuration records db.client.operation.duration metric for pipelines/transactions.
// operationName should be "PIPELINE" for regular pipelines or "MULTI" for transactions.
func (r *metricsRecorder) RecordPipelineOperationDuration(
	ctx context.Context,
	duration time.Duration,
	operationName string,
	cmdCount int,
	attempts int,
	err error,
	cn redis.ConnInfo,
	dbIndex int,
) {
	if r.operationDuration == nil {
		return
	}

	// Convert duration to seconds (OTel convention for duration metrics)
	durationSeconds := duration.Seconds()

	// Extract server info from connection
	serverAddr, serverPort := extractServerInfo(cn)

	// Build attributes
	attrs := []attribute.KeyValue{
		attribute.String(AttrDBOperationName, operationName),
		getLibraryVersionAttr(),
		attribute.Int(AttrRedisClientOperationRetryAttempts, attempts-1), // attempts-1 = retry count
		attribute.Int(AttrDBOperationBatchSize, cmdCount),                // number of commands in pipeline
		attribute.String(AttrDBSystemName, DBSystemRedis),
		attribute.String(AttrServerAddress, serverAddr),
		attribute.String(AttrDBNamespace, strconv.Itoa(dbIndex)),
	}

	// Add server.port if not default
	attrs = addServerPortIfNonDefault(attrs, serverPort)

	// Add network.peer.address and network.peer.port from connection
	if cn != nil {
		remoteAddr := cn.RemoteAddr()
		if remoteAddr != nil {
			peerAddr, peerPort := splitHostPort(remoteAddr.String())
			if peerAddr != "" {
				attrs = append(attrs, attribute.String(AttrNetworkPeerAddress, peerAddr))
			}
			if peerPort != "" {
				attrs = append(attrs, attribute.String(AttrNetworkPeerPort, peerPort))
			}
		}
	}

	// Add error attributes if pipeline failed
	if err != nil {
		attrs = append(attrs, attribute.String(AttrErrorType, classifyError(err)))
		attrs = append(attrs, attribute.String(AttrRedisClientErrorsCategory, getErrorCategory(err)))
		if statusCode := extractRedisErrorPrefix(err); statusCode != "" {
			attrs = append(attrs, attribute.String(AttrDBResponseStatusCode, statusCode))
		}
	}

	// Record the histogram
	r.operationDuration.Record(ctx, durationSeconds, metric.WithAttributes(attrs...))
}

// classifyError returns the error.type attribute value
// Format: <category>:<subcategory>:<error_name>
func classifyError(err error) string {
	if err == nil {
		return ""
	}

	// Timeout errors
	if isTimeoutError(err) {
		return "timeout"
	}

	// Network errors - normalize to avoid high cardinality
	if isNetworkError(err) {
		return normalizeNetworkError(err)
	}

	// Redis errors (start with error prefix like ERR, WRONGTYPE, etc.)
	if prefix := extractRedisErrorPrefix(err); prefix != "" {
		return fmt.Sprintf("redis:%s", prefix)
	}

	// Generic error - normalize to avoid high cardinality
	return normalizeGenericError(err.Error())
}

// normalizeNetworkError normalizes network error messages to prevent high cardinality
// by removing variable data like port numbers and connection details
func normalizeNetworkError(err error) string {
	if err == nil {
		return "network:unknown"
	}

	errStr := strings.ToLower(err.Error())

	// I/O timeout errors
	if strings.Contains(errStr, "i/o timeout") {
		return "network:io_timeout"
	}

	// Connection refused errors
	if strings.Contains(errStr, "connection refused") {
		return "network:connection_refused"
	}

	// Connection reset errors
	if strings.Contains(errStr, "connection reset") {
		return "network:connection_reset"
	}

	// EOF errors
	if strings.Contains(errStr, "eof") {
		return "network:eof"
	}

	// Broken pipe
	if strings.Contains(errStr, "broken pipe") {
		return "network:broken_pipe"
	}

	// Connection closed
	if strings.Contains(errStr, "use of closed") || strings.Contains(errStr, "connection closed") {
		return "network:connection_closed"
	}

	// DNS/hostname errors
	if strings.Contains(errStr, "no such host") || strings.Contains(errStr, "lookup") {
		return "network:dns_error"
	}

	// Default to generic network error
	return "network:other"
}

// normalizeGenericError normalizes generic error messages to prevent high cardinality
func normalizeGenericError(errStr string) string {
	// Remove common variable patterns
	// Port numbers, IP addresses, etc. would create high cardinality
	errLower := strings.ToLower(errStr)

	// Context deadline exceeded
	if strings.Contains(errLower, "context deadline exceeded") {
		return "context_deadline_exceeded"
	}

	// Context canceled
	if strings.Contains(errLower, "context canceled") {
		return "context_canceled"
	}

	// Pool errors
	if strings.Contains(errLower, "pool") {
		if strings.Contains(errLower, "timeout") {
			return "pool_timeout"
		}
		if strings.Contains(errLower, "closed") {
			return "pool_closed"
		}
		return "pool_error"
	}

	// Return first word if it looks like a category, otherwise return "other"
	parts := strings.SplitN(errStr, " ", 2)
	if len(parts) > 0 && len(parts[0]) > 0 && len(parts[0]) <= 30 {
		// Only use the first word if it doesn't contain variable data
		firstWord := strings.TrimSuffix(strings.TrimSuffix(parts[0], ":"), ",")
		if !strings.ContainsAny(firstWord, "0123456789./[]") {
			return strings.ToLower(firstWord)
		}
	}

	return "other"
}

// extractRedisErrorPrefix extracts the Redis error prefix (e.g., "ERR", "WRONGTYPE")
// Redis errors typically start with an uppercase prefix followed by a space
func extractRedisErrorPrefix(err error) string {
	if err == nil {
		return ""
	}

	errStr := err.Error()

	// Redis errors typically start with an uppercase prefix
	// Examples: "ERR ...", "WRONGTYPE ...", "CLUSTERDOWN ..."
	parts := strings.SplitN(errStr, " ", 2)
	if len(parts) > 0 {
		prefix := parts[0]
		// Check if it's all uppercase (Redis error convention)
		if prefix == strings.ToUpper(prefix) && len(prefix) > 0 {
			return prefix
		}
	}

	return ""
}

// isNetworkError checks if an error is a network-related error
func isNetworkError(err error) bool {
	if err == nil {
		return false
	}

	// Check for net.Error interface (standard way to detect network errors)
	_, ok := err.(net.Error)
	return ok
}

// isTimeoutError checks if an error is a timeout error
func isTimeoutError(err error) bool {
	if err == nil {
		return false
	}

	// Check for timeoutError interface (works with wrapped errors)
	var te timeoutError
	if errors.As(err, &te) {
		return te.Timeout()
	}

	// Check for net.Error specifically (common case for network timeouts)
	var netErr net.Error
	if errors.As(err, &netErr) {
		return netErr.Timeout()
	}

	return false
}

// getErrorCategory returns the error category from an error.
// It checks for TLS, auth, network, and server errors.
func getErrorCategory(err error) string {
	if err == nil {
		return ""
	}

	errStr := err.Error()

	// For actual errors, also check error types
	if isNetworkError(err) || isTimeoutError(err) {
		return "network"
	}

	// Check for Redis server errors by prefix
	if prefix := extractRedisErrorPrefix(err); prefix != "" {
		return "server"
	}

	return getErrorCategoryFromString(errStr)
}

// getErrorCategoryFromString returns the error category from an error string or error type.
// This is used both by getErrorCategory (for errors) and directly for error type strings.
func getErrorCategoryFromString(errStr string) string {
	if errStr == "" {
		return ""
	}

	errLower := strings.ToLower(errStr)

	// Check for TLS errors
	if strings.Contains(errLower, "tls") || strings.Contains(errLower, "certificate") ||
		strings.Contains(errLower, "x509") || strings.Contains(errLower, "ssl") {
		return "tls"
	}

	// Check for auth errors
	if strings.Contains(errLower, "auth") || strings.Contains(errLower, "noauth") ||
		strings.Contains(errLower, "wrongpass") || strings.Contains(errLower, "noperm") ||
		strings.Contains(errLower, "permission") {
		return "auth"
	}

	// Check for network errors (transport/DNS/socket issues)
	if strings.Contains(errLower, "network") || strings.Contains(errLower, "timeout") ||
		strings.Contains(errLower, "connection refused") ||
		strings.Contains(errLower, "connection reset") ||
		strings.Contains(errLower, "no route to host") ||
		strings.Contains(errLower, "network is unreachable") ||
		strings.Contains(errLower, "dns lookup") ||
		strings.Contains(errLower, "dial") ||
		strings.Contains(errLower, "eof") ||
		strings.Contains(errLower, "broken pipe") ||
		strings.Contains(errLower, "i/o") {
		return "network"
	}

	// Check for Redis server errors (including cluster errors)
	// Common Redis error prefixes: ERR, WRONGTYPE, CLUSTERDOWN, MOVED, ASK, READONLY, etc.
	serverErrors := []string{"err", "wrongtype", "clusterdown", "moved", "ask", "readonly",
		"crossslot", "tryagain", "loading", "busy", "noscript", "oom", "execabort", "noquorum", "redis:"}
	for _, prefix := range serverErrors {
		if strings.Contains(errLower, prefix) {
			return "server"
		}
	}

	// If it looks like an uppercase Redis error prefix, it's a server error
	if len(errStr) > 0 && errStr == strings.ToUpper(errStr) {
		return "server"
	}

	// Uncategorized errors
	return "other"
}

// splitHostPort splits a host:port string into host and port
// This is a simplified version that handles the common cases
func splitHostPort(addr string) (host, port string) {
	// Handle Unix sockets
	if strings.HasPrefix(addr, "/") || strings.HasPrefix(addr, "@") {
		return addr, ""
	}

	host, port, err := net.SplitHostPort(addr)
	if err != nil {
		// If split fails, return the whole address as host
		return addr, ""
	}

	return host, port
}

// parseAddr parses a Redis address into host and port
func parseAddr(addr string) (host, port string) {
	// Handle Unix sockets
	if strings.HasPrefix(addr, "/") || strings.HasPrefix(addr, "unix://") {
		return addr, ""
	}

	// Remove protocol prefix if present
	addr = strings.TrimPrefix(addr, "redis://")
	addr = strings.TrimPrefix(addr, "rediss://")

	host, port, err := net.SplitHostPort(addr)
	if err != nil {
		// No port specified, use default
		return addr, "6379"
	}

	return host, port
}

// extractServerInfo extracts server address and port from connection info
// For client connections, this is the remote endpoint (server address)
func extractServerInfo(cn redis.ConnInfo) (addr, port string) {
	if cn == nil {
		return "", ""
	}

	remoteAddr := cn.RemoteAddr()
	if remoteAddr == nil {
		return "", ""
	}

	addrStr := remoteAddr.String()
	host, portStr := parseAddr(addrStr)
	return host, portStr
}

// RecordConnectionCreateTime records the time it took to create a new connection
func (r *metricsRecorder) RecordConnectionCreateTime(
	ctx context.Context,
	duration time.Duration,
	cn redis.ConnInfo,
) {
	if r.connectionCreateTime == nil {
		return
	}

	// Convert duration to seconds (OTel convention)
	durationSeconds := duration.Seconds()

	// Build attributes
	attrs := []attribute.KeyValue{
		attribute.String(AttrDBSystemName, DBSystemRedis),
		getLibraryVersionAttr(),
	}

	// Use pool name from connection (set when connection was created)
	// This ensures consistency with gauge metrics which also use the registered pool name
	if cn != nil {
		poolName := cn.PoolName()
		if poolName != "" {
			attrs = append(attrs, attribute.String(AttrDBClientConnectionPoolName, poolName))
		}
	}

	// Record the histogram
	r.connectionCreateTime.Record(ctx, durationSeconds, metric.WithAttributes(attrs...))
}

// RecordConnectionRelaxedTimeout records when connection timeout is relaxed/unrelaxed
func (r *metricsRecorder) RecordConnectionRelaxedTimeout(
	ctx context.Context,
	delta int,
	cn redis.ConnInfo,
	poolName, notificationType string,
) {
	if r.connectionRelaxedTimeout == nil {
		return
	}

	// Build attributes
	attrs := []attribute.KeyValue{
		attribute.String(AttrDBSystemName, DBSystemRedis),
		getLibraryVersionAttr(),
	}

	// Use pool name from connection (set when connection was created)
	if cn != nil {
		connPoolName := cn.PoolName()
		if connPoolName != "" {
			attrs = append(attrs, attribute.String(AttrDBClientConnectionPoolName, connPoolName))
		}
	}

	// Add notification type
	attrs = append(attrs, attribute.String(AttrRedisClientConnectionNotification, notificationType))

	// Record the counter (delta can be +1 or -1)
	r.connectionRelaxedTimeout.Add(ctx, int64(delta), metric.WithAttributes(attrs...))
}

// RecordConnectionHandoff records when a connection is handed off to another node
func (r *metricsRecorder) RecordConnectionHandoff(
	ctx context.Context,
	cn redis.ConnInfo,
	poolName string,
) {
	if r.connectionHandoff == nil {
		return
	}

	attrs := []attribute.KeyValue{
		attribute.String(AttrDBSystemName, DBSystemRedis),
		getLibraryVersionAttr(),
	}

	// Use pool name from connection (set when connection was created)
	if cn != nil {
		connPoolName := cn.PoolName()
		if connPoolName != "" {
			attrs = append(attrs, attribute.String(AttrDBClientConnectionPoolName, connPoolName))
		}
	}

	// Record the counter
	r.connectionHandoff.Add(ctx, 1, metric.WithAttributes(attrs...))
}

// RecordError records client errors (ASK, MOVED, handshake failures, etc.)
func (r *metricsRecorder) RecordError(
	ctx context.Context,
	errorType string,
	cn redis.ConnInfo,
	statusCode string,
	isInternal bool,
	retryAttempts int,
) {
	if r.clientErrors == nil {
		return
	}

	// Extract server address and peer address from connection (may be nil for some errors)
	// For client connections, peer address is the same as server address (remote endpoint)
	var serverAddr, serverPort, peerAddr, peerPort string
	if cn != nil {
		serverAddr, serverPort = extractServerInfo(cn)
		peerAddr, peerPort = serverAddr, serverPort // Peer is same as server for client connections
	}

	// Build attributes
	attrs := []attribute.KeyValue{
		attribute.String(AttrDBSystemName, DBSystemRedis),
		attribute.String(AttrErrorType, errorType),
		attribute.String(AttrRedisClientErrorsCategory, getErrorCategoryFromString(errorType)),
		attribute.String(AttrDBResponseStatusCode, statusCode),
		attribute.Bool(AttrRedisClientErrorsInternal, isInternal),
		attribute.Int(AttrRedisClientOperationRetryAttempts, retryAttempts),
		getLibraryVersionAttr(),
	}

	// Add server info if available
	if serverAddr != "" {
		attrs = append(attrs, attribute.String(AttrServerAddress, serverAddr))
		attrs = addServerPortIfNonDefault(attrs, serverPort)
	}

	// Add peer info if available
	if peerAddr != "" {
		attrs = append(attrs, attribute.String(AttrNetworkPeerAddress, peerAddr))
		if peerPort != "" {
			attrs = append(attrs, attribute.String(AttrNetworkPeerPort, peerPort))
		}
	}

	// Record the counter
	r.clientErrors.Add(ctx, 1, metric.WithAttributes(attrs...))
}

// RecordMaintenanceNotification records when a maintenance notification is received
func (r *metricsRecorder) RecordMaintenanceNotification(
	ctx context.Context,
	cn redis.ConnInfo,
	notificationType string,
) {
	if r.maintenanceNotifications == nil {
		return
	}

	// Extract server address and peer address from connection
	// For client connections, peer address is the same as server address (remote endpoint)
	serverAddr, serverPort := extractServerInfo(cn)
	peerAddr, peerPort := serverAddr, serverPort // Peer is same as server for client connections

	// Build attributes
	attrs := []attribute.KeyValue{
		attribute.String(AttrDBSystemName, DBSystemRedis),
		attribute.String(AttrServerAddress, serverAddr),
		getLibraryVersionAttr(),
		attribute.String(AttrRedisClientConnectionNotification, notificationType),
	}

	// Add server.port if not default
	attrs = addServerPortIfNonDefault(attrs, serverPort)

	// Add peer info if available
	if peerAddr != "" {
		attrs = append(attrs, attribute.String(AttrNetworkPeerAddress, peerAddr))
		if peerPort != "" {
			attrs = append(attrs, attribute.String(AttrNetworkPeerPort, peerPort))
		}
	}

	// Record the counter
	r.maintenanceNotifications.Add(ctx, 1, metric.WithAttributes(attrs...))
}

// RecordConnectionWaitTime records db.client.connection.wait_time metric
func (r *metricsRecorder) RecordConnectionWaitTime(
	ctx context.Context,
	duration time.Duration,
	cn redis.ConnInfo,
) {
	if r.connectionWaitTime == nil {
		return
	}

	// Build attributes
	attrs := []attribute.KeyValue{
		attribute.String(AttrDBSystemName, DBSystemRedis),
		getLibraryVersionAttr(),
	}

	// 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))
		}
	}

	// Record the histogram (duration in seconds)
	r.connectionWaitTime.Record(ctx, duration.Seconds(), metric.WithAttributes(attrs...))
}

// RecordConnectionClosed records redis.client.connection.closed metric
func (r *metricsRecorder) RecordConnectionClosed(
	ctx context.Context,
	cn redis.ConnInfo,
	reason string,
	err error,
) {
	if r.connectionClosed == nil {
		return
	}

	// Build attributes
	attrs := []attribute.KeyValue{
		attribute.String(AttrDBSystemName, DBSystemRedis),
		getLibraryVersionAttr(),
	}

	// 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))
		}
	}

	// Add error type and category (always required per spec)
	// Use classifyError to normalize error messages and prevent high cardinality
	if err != nil {
		// Normalize the close reason to prevent high cardinality from variable data
		// (e.g., port numbers, connection IDs in error messages)
		normalizedReason := classifyError(err)
		attrs = append(attrs, attribute.String(AttrRedisClientConnectionCloseReason, normalizedReason))
		attrs = append(attrs, attribute.String(AttrErrorType, normalizedReason))
		attrs = append(attrs, attribute.String(AttrRedisClientErrorsCategory, getErrorCategory(err)))
	} else {
		// For non-error closures, use reason directly (these are controlled strings like "pool_closed")
		attrs = append(attrs, attribute.String(AttrRedisClientConnectionCloseReason, reason))
		attrs = append(attrs, attribute.String(AttrErrorType, reason))
		attrs = append(attrs, attribute.String(AttrRedisClientErrorsCategory, getErrorCategoryFromString(reason)))
	}

	// Record the counter
	r.connectionClosed.Add(ctx, 1, metric.WithAttributes(attrs...))
}

// RecordPubSubMessage records redis.client.pubsub.messages metric
func (r *metricsRecorder) RecordPubSubMessage(
	ctx context.Context,
	cn redis.ConnInfo,
	direction string,
	channel string,
	sharded bool,
) {
	if r.pubsubMessages == nil {
		return
	}

	// Extract server address and peer address from connection
	serverAddr, serverPort := extractServerInfo(cn)
	peerAddr, peerPort := serverAddr, serverPort

	// Build attributes
	attrs := []attribute.KeyValue{
		attribute.String(AttrDBSystemName, DBSystemRedis),
		attribute.String(AttrServerAddress, serverAddr),
		attribute.String(AttrRedisClientPubSubDirection, direction), // "sent" or "received"
		attribute.Bool(AttrRedisClientPubSubSharded, sharded),
		getLibraryVersionAttr(),
	}

	// Add channel name if not hidden for cardinality reduction
	if !r.cfg.hidePubSubChannelNames && channel != "" {
		attrs = append(attrs, attribute.String(AttrRedisClientPubSubChannel, channel))
	}

	// Add server.port if not default
	attrs = addServerPortIfNonDefault(attrs, serverPort)

	// Add peer info
	if peerAddr != "" {
		attrs = append(attrs, attribute.String(AttrNetworkPeerAddress, peerAddr))
		if peerPort != "" {
			attrs = append(attrs, attribute.String(AttrNetworkPeerPort, peerPort))
		}
	}

	// Record the counter
	r.pubsubMessages.Add(ctx, 1, metric.WithAttributes(attrs...))
}

// RecordStreamLag records redis.client.stream.lag metric
func (r *metricsRecorder) RecordStreamLag(
	ctx context.Context,
	lag time.Duration,
	cn redis.ConnInfo,
	streamName string,
	consumerGroup string,
	consumerName string,
) {
	if r.streamLag == nil {
		return
	}

	// Extract server address and peer address from connection
	serverAddr, serverPort := extractServerInfo(cn)
	peerAddr, peerPort := serverAddr, serverPort

	// Build attributes
	attrs := []attribute.KeyValue{
		attribute.String(AttrDBSystemName, DBSystemRedis),
		attribute.String(AttrServerAddress, serverAddr),
		attribute.String(AttrRedisClientStreamConsumerGroup, consumerGroup),
		attribute.String(AttrRedisClientStreamConsumerName, consumerName),
		getLibraryVersionAttr(),
	}

	// Add stream name if not hidden for cardinality reduction
	if !r.cfg.hideStreamNames && streamName != "" {
		attrs = append(attrs, attribute.String(AttrRedisClientStreamName, streamName))
	}

	// Add server.port if not default
	attrs = addServerPortIfNonDefault(attrs, serverPort)

	// Add peer info
	if peerAddr != "" {
		attrs = append(attrs, attribute.String(AttrNetworkPeerAddress, peerAddr))
		if peerPort != "" {
			attrs = append(attrs, attribute.String(AttrNetworkPeerPort, peerPort))
		}
	}

	// Record the histogram (lag in seconds)
	r.streamLag.Record(ctx, lag.Seconds(), metric.WithAttributes(attrs...))
}

// 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
	}

	// Build attributes
	attrs := []attribute.KeyValue{
		attribute.String(AttrDBSystemName, DBSystemRedis),
		getLibraryVersionAttr(),
		attribute.String(AttrDBClientConnectionState, state),
		attribute.Bool(AttrRedisClientConnectionPubSub, isPubSub),
	}

	// 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))

			// 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...))
}

// 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...))
}