Add phase 1 metrics

Author: ofekshenawa

extra/redisotel-native/metrics.go

@@ -20,9 +20,13 @@ const (
 
 // metricsRecorder implements the otel.Recorder interface
 type metricsRecorder struct {
-	operationDuration    metric.Float64Histogram
-	connectionCount      metric.Int64UpDownCounter
-	connectionCreateTime metric.Float64Histogram
+	operationDuration        metric.Float64Histogram
+	connectionCount          metric.Int64UpDownCounter
+	connectionCreateTime     metric.Float64Histogram
+	connectionRelaxedTimeout metric.Int64UpDownCounter
+	connectionHandoff        metric.Int64Counter
+	clientErrors             metric.Int64Counter
+	maintenanceNotifications metric.Int64Counter
 
 	// Client configuration for attributes (used for operation metrics only)
 	serverAddr string
@@ -54,7 +58,7 @@ func (r *metricsRecorder) RecordOperationDuration(
 		attribute.Bool("redis.client.operation.blocking", isBlockingCommand(cmd)),
 
 		// Recommended attributes
-		attribute.String("db.system", "redis"),
+		attribute.String("db.system.name", "redis"),
 		attribute.String("server.address", r.serverAddr),
 	}
 
@@ -326,6 +330,12 @@ func extractServerInfo(cn redis.ConnInfo) (addr, port string) {
 	return host, portStr
 }
 
+// extractPeerInfo extracts peer network address and port from connection info
+// For client connections, this is the same as the server address (remote endpoint)
+func extractPeerInfo(cn redis.ConnInfo) (addr, port string) {
+	return extractServerInfo(cn)
+}
+
 // RecordConnectionCreateTime records the time it took to create a new connection
 func (r *metricsRecorder) RecordConnectionCreateTime(
 	ctx context.Context,
@@ -364,3 +374,154 @@ func (r *metricsRecorder) RecordConnectionCreateTime(
 	// 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
+	}
+
+	// Extract server address from connection
+	serverAddr, serverPort := extractServerInfo(cn)
+
+	// Build attributes
+	attrs := []attribute.KeyValue{
+		attribute.String("db.system.name", "redis"),
+		attribute.String("server.address", serverAddr),
+		attribute.String("redis.client.library", fmt.Sprintf("%s:%s", libraryName, redis.Version())),
+		attribute.String("db.client.connection.pool.name", poolName),
+		attribute.String("redis.client.connection.notification", notificationType),
+	}
+
+	// Add server.port if not default
+	if serverPort != "" && serverPort != "6379" {
+		attrs = append(attrs, attribute.String("server.port", serverPort))
+	}
+
+	// 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
+	}
+
+	// Extract server address from connection
+	serverAddr, serverPort := extractServerInfo(cn)
+
+	// Build attributes
+	attrs := []attribute.KeyValue{
+		attribute.String("db.system", "redis"),
+		attribute.String("server.address", serverAddr),
+		attribute.String("redis.client.library", fmt.Sprintf("%s:%s", libraryName, redis.Version())),
+		attribute.String("db.client.connection.pool.name", poolName),
+	}
+
+	// Add server.port if not default
+	if serverPort != "" && serverPort != "6379" {
+		attrs = append(attrs, attribute.String("server.port", serverPort))
+	}
+
+	// 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)
+	var serverAddr, serverPort, peerAddr, peerPort string
+	if cn != nil {
+		serverAddr, serverPort = extractServerInfo(cn)
+		peerAddr, peerPort = extractPeerInfo(cn)
+	}
+
+	// Build attributes
+	attrs := []attribute.KeyValue{
+		attribute.String("db.system.name", "redis"),
+		attribute.String("error.type", errorType),
+		attribute.String("db.response.status_code", statusCode),
+		attribute.Bool("redis.client.errors.internal", isInternal),
+		attribute.Int("redis.client.operation.retry_attempts", retryAttempts),
+		attribute.String("redis.client.library", fmt.Sprintf("%s:%s", libraryName, redis.Version())),
+	}
+
+	// Add server info if available
+	if serverAddr != "" {
+		attrs = append(attrs, attribute.String("server.address", serverAddr))
+		if serverPort != "" && serverPort != "6379" {
+			attrs = append(attrs, attribute.String("server.port", serverPort))
+		}
+	}
+
+	// Add peer info if available
+	if peerAddr != "" {
+		attrs = append(attrs, attribute.String("network.peer.address", peerAddr))
+		if peerPort != "" {
+			attrs = append(attrs, attribute.String("network.peer.port", 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
+	serverAddr, serverPort := extractServerInfo(cn)
+	peerAddr, peerPort := extractPeerInfo(cn)
+
+	// Build attributes
+	attrs := []attribute.KeyValue{
+		attribute.String("db.system.name", "redis"),
+		attribute.String("server.address", serverAddr),
+		attribute.String("redis.client.library", fmt.Sprintf("%s:%s", libraryName, redis.Version())),
+		attribute.String("redis.client.connection.notification", notificationType),
+	}
+
+	// Add server.port if not default
+	if serverPort != "" && serverPort != "6379" {
+		attrs = append(attrs, attribute.String("server.port", serverPort))
+	}
+
+	// Add peer info if available
+	if peerAddr != "" {
+		attrs = append(attrs, attribute.String("network.peer.address", peerAddr))
+		if peerPort != "" {
+			attrs = append(attrs, attribute.String("network.peer.port", peerPort))
+		}
+	}
+
+	// Record the counter
+	r.maintenanceNotifications.Add(ctx, 1, metric.WithAttributes(attrs...))
+}

extra/redisotel-native/redisotel.go

@@ -145,14 +145,57 @@ func initOnce(client redis.UniversalClient, opts ...Option) error {
 		return fmt.Errorf("failed to create connection create time histogram: %w", err)
 	}
 
+	// Create UpDownCounter for relaxed timeout tracking
+	connectionRelaxedTimeout, err := meter.Int64UpDownCounter(
+		"redis.client.connection.relaxed_timeout",
+		metric.WithDescription("How many times the connection timeout has been increased/decreased (after a server maintenance notification)"),
+		metric.WithUnit("{relaxation}"),
+	)
+	if err != nil {
+		return fmt.Errorf("failed to create connection relaxed timeout metric: %w", err)
+	}
+
+	// Create Counter for connection handoffs
+	connectionHandoff, err := meter.Int64Counter(
+		"redis.client.connection.handoff",
+		metric.WithDescription("Connections that have been handed off to another node (e.g after a MOVING notification)"),
+	)
+	if err != nil {
+		return fmt.Errorf("failed to create connection handoff metric: %w", err)
+	}
+
+	// Create Counter for client errors
+	clientErrors, err := meter.Int64Counter(
+		"redis.client.errors",
+		metric.WithDescription("Number of errors handled by the Redis client"),
+		metric.WithUnit("{error}"),
+	)
+	if err != nil {
+		return fmt.Errorf("failed to create client errors metric: %w", err)
+	}
+
+	// Create Counter for maintenance notifications
+	maintenanceNotifications, err := meter.Int64Counter(
+		"redis.client.maintenance.notifications",
+		metric.WithDescription("Number of maintenance notifications received"),
+		metric.WithUnit("{notification}"),
+	)
+	if err != nil {
+		return fmt.Errorf("failed to create maintenance notifications metric: %w", err)
+	}
+
 	// Create recorder
 	recorder := &metricsRecorder{
-		operationDuration:    operationDuration,
-		connectionCount:      connectionCount,
-		connectionCreateTime: connectionCreateTime,
-		serverAddr:           serverAddr,
-		serverPort:           serverPort,
-		dbIndex:              dbIndex,
+		operationDuration:        operationDuration,
+		connectionCount:          connectionCount,
+		connectionCreateTime:     connectionCreateTime,
+		connectionRelaxedTimeout: connectionRelaxedTimeout,
+		connectionHandoff:        connectionHandoff,
+		clientErrors:             clientErrors,
+		maintenanceNotifications: maintenanceNotifications,
+		serverAddr:               serverAddr,
+		serverPort:               serverPort,
+		dbIndex:                  dbIndex,
 	}
 
 	// Register global recorder

internal/otel/metrics.go

@@ -27,6 +27,27 @@ type Recorder interface {
 
 	// RecordConnectionCreateTime records the time it took to create a new connection
 	RecordConnectionCreateTime(ctx context.Context, duration time.Duration, cn *pool.Conn)
+
+	// RecordConnectionRelaxedTimeout records when connection timeout is relaxed/unrelaxed
+	// delta: +1 for relaxed, -1 for unrelaxed
+	// poolName: name of the connection pool (e.g., "main", "pubsub")
+	// notificationType: the notification type that triggered the timeout relaxation (e.g., "MOVING")
+	RecordConnectionRelaxedTimeout(ctx context.Context, delta int, cn *pool.Conn, poolName, notificationType string)
+
+	// RecordConnectionHandoff records when a connection is handed off to another node
+	// poolName: name of the connection pool (e.g., "main", "pubsub")
+	RecordConnectionHandoff(ctx context.Context, cn *pool.Conn, poolName string)
+
+	// RecordError records client errors (ASK, MOVED, handshake failures, etc.)
+	// errorType: type of error (e.g., "ASK", "MOVED", "HANDSHAKE_FAILED")
+	// statusCode: Redis response status code if available (e.g., "MOVED", "ASK")
+	// isInternal: whether this is an internal error
+	// retryAttempts: number of retry attempts made
+	RecordError(ctx context.Context, errorType string, cn *pool.Conn, statusCode string, isInternal bool, retryAttempts int)
+
+	// RecordMaintenanceNotification records when a maintenance notification is received
+	// notificationType: the type of notification (e.g., "MOVING", "MIGRATING", etc.)
+	RecordMaintenanceNotification(ctx context.Context, cn *pool.Conn, notificationType string)
 }
 
 // Global recorder instance (initialized by extra/redisotel-native)
@@ -39,6 +60,10 @@ func SetGlobalRecorder(r Recorder) {
 		// Unregister pool callbacks
 		pool.SetConnectionStateChangeCallback(nil)
 		pool.SetConnectionCreateTimeCallback(nil)
+		pool.SetConnectionRelaxedTimeoutCallback(nil)
+		pool.SetConnectionHandoffCallback(nil)
+		pool.SetErrorCallback(nil)
+		pool.SetMaintenanceNotificationCallback(nil)
 		return
 	}
 	globalRecorder = r
@@ -52,6 +77,26 @@ func SetGlobalRecorder(r Recorder) {
 	pool.SetConnectionCreateTimeCallback(func(ctx context.Context, duration time.Duration, cn *pool.Conn) {
 		globalRecorder.RecordConnectionCreateTime(ctx, duration, cn)
 	})
+
+	// Register pool callback to forward connection relaxed timeout changes to recorder
+	pool.SetConnectionRelaxedTimeoutCallback(func(ctx context.Context, delta int, cn *pool.Conn, poolName, notificationType string) {
+		globalRecorder.RecordConnectionRelaxedTimeout(ctx, delta, cn, poolName, notificationType)
+	})
+
+	// Register pool callback to forward connection handoffs to recorder
+	pool.SetConnectionHandoffCallback(func(ctx context.Context, cn *pool.Conn, poolName string) {
+		globalRecorder.RecordConnectionHandoff(ctx, cn, poolName)
+	})
+
+	// Register pool callback to forward errors to recorder
+	pool.SetErrorCallback(func(ctx context.Context, errorType string, cn *pool.Conn, statusCode string, isInternal bool, retryAttempts int) {
+		globalRecorder.RecordError(ctx, errorType, cn, statusCode, isInternal, retryAttempts)
+	})
+
+	// Register pool callback to forward maintenance notifications to recorder
+	pool.SetMaintenanceNotificationCallback(func(ctx context.Context, cn *pool.Conn, notificationType string) {
+		globalRecorder.RecordMaintenanceNotification(ctx, cn, notificationType)
+	})
 }
 
 // RecordOperationDuration records the total operation duration.
@@ -76,5 +121,10 @@ func RecordConnectionCreateTime(ctx context.Context, duration time.Duration, cn
 type noopRecorder struct{}
 
 func (noopRecorder) RecordOperationDuration(context.Context, time.Duration, Cmder, int, *pool.Conn) {}
-func (noopRecorder) RecordConnectionStateChange(context.Context, *pool.Conn, string, string)       {}
-func (noopRecorder) RecordConnectionCreateTime(context.Context, time.Duration, *pool.Conn)         {}
+func (noopRecorder) RecordConnectionStateChange(context.Context, *pool.Conn, string, string)        {}
+func (noopRecorder) RecordConnectionCreateTime(context.Context, time.Duration, *pool.Conn)          {}
+func (noopRecorder) RecordConnectionRelaxedTimeout(context.Context, int, *pool.Conn, string, string) {
+}
+func (noopRecorder) RecordConnectionHandoff(context.Context, *pool.Conn, string)        {}
+func (noopRecorder) RecordError(context.Context, string, *pool.Conn, string, bool, int) {}
+func (noopRecorder) RecordMaintenanceNotification(context.Context, *pool.Conn, string)  {}

internal/pool/conn.go

@@ -241,6 +241,8 @@ func (cn *Conn) SetUsable(usable bool) {
 // SetRelaxedTimeout sets relaxed timeouts for this connection during maintenanceNotifications upgrades.
 // These timeouts will be used for all subsequent commands until the deadline expires.
 // Uses atomic operations for lock-free access.
+// Note: Metrics should be recorded by the caller (notification handler) which has context about
+// the notification type and pool name.
 func (cn *Conn) SetRelaxedTimeout(readTimeout, writeTimeout time.Duration) {
 	cn.relaxedCounter.Add(1)
 	cn.relaxedReadTimeoutNs.Store(int64(readTimeout))
@@ -275,6 +277,11 @@ func (cn *Conn) clearRelaxedTimeout() {
 	cn.relaxedWriteTimeoutNs.Store(0)
 	cn.relaxedDeadlineNs.Store(0)
 	cn.relaxedCounter.Store(0)
+
+	// Note: Metrics for timeout unrelaxing are not recorded here because we don't have
+	// context about which notification type or pool triggered the relaxation.
+	// In practice, relaxed timeouts expire automatically via deadline, so explicit
+	// unrelaxing metrics are less critical than the initial relaxation metrics.
 }
 
 // HasRelaxedTimeout returns true if relaxed timeouts are currently active on this connection.