improve logging

Author: ndyakov

hitless/config.go

@@ -1,11 +1,13 @@
 package hitless
 
 import (
+	"context"
 	"net"
 	"runtime"
 	"strings"
 	"time"
 
+	"github.com/redis/go-redis/v9/internal"
 	"github.com/redis/go-redis/v9/internal/util"
 )
 
@@ -279,6 +281,11 @@ func (c *Config) ApplyDefaultsWithPoolSize(poolSize int) *Config {
 		result.MaxHandoffRetries = c.MaxHandoffRetries
 	}
 
+	if result.LogLevel >= 3 {
+		result.LogLevel = 3
+		internal.Logger.Printf(context.Background(), "hitless: debug logging enabled")
+		internal.Logger.Printf(context.Background(), "hitless: config: %+v", result)
+	}
 	return result
 }
 

hitless/hitless_manager.go

@@ -148,7 +148,9 @@ func (hm *HitlessManager) TrackMovingOperationWithConnID(ctx context.Context, ne
 	// Use LoadOrStore for atomic check-and-set operation
 	if _, loaded := hm.activeMovingOps.LoadOrStore(key, movingOp); loaded {
 		// Duplicate MOVING notification, ignore
-		internal.Logger.Printf(ctx, "Duplicate MOVING operation ignored: %s", key.String())
+		if hm.config.LogLevel >= 3 { // Warning level
+			internal.Logger.Printf(ctx, "hitless: conn[%d] seqID[%d] Duplicate MOVING operation ignored: %s", connID, seqID, key.String())
+		}
 		return nil
 	}
 
@@ -170,6 +172,10 @@ func (hm *HitlessManager) UntrackOperationWithConnID(seqID int64, connID uint64)
 	if _, loaded := hm.activeMovingOps.LoadAndDelete(key); loaded {
 		// Decrement active operation count only if operation existed
 		hm.activeOperationCount.Add(-1)
+	} else {
+		if hm.config.LogLevel >= 3 { // Warning level
+			internal.Logger.Printf(context.Background(), "hitless: conn[%d] seqID[%d] Operation not found for untracking: %s", connID, seqID, key.String())
+		}
 	}
 }
 

hitless/notification_handler.go

@@ -6,7 +6,6 @@ import (
 	"time"
 
 	"github.com/redis/go-redis/v9/internal"
-	"github.com/redis/go-redis/v9/internal/interfaces"
 	"github.com/redis/go-redis/v9/internal/pool"
 	"github.com/redis/go-redis/v9/push"
 )
@@ -107,6 +106,10 @@ func (snh *NotificationHandler) handleMoving(ctx context.Context, handlerCtx pus
 	deadline := time.Now().Add(time.Duration(timeS) * time.Second)
 	// If newEndpoint is empty, we should schedule a handoff to the current endpoint in timeS/2 seconds
 	if newEndpoint == "" || newEndpoint == internal.RedisNull {
+		if snh.manager.config.LogLevel >= 3 { // Debug level
+			internal.Logger.Printf(ctx, "hitless: conn[%d] scheduling handoff to current endpoint in %v seconds",
+				poolConn.GetID(), timeS/2)
+		}
 		// same as current endpoint
 		newEndpoint = snh.manager.options.GetAddr()
 		// delay the handoff for timeS/2 seconds to the same endpoint
@@ -137,7 +140,6 @@ func (snh *NotificationHandler) markConnForHandoff(conn *pool.Conn, newEndpoint
 	// Optionally track in hitless manager for monitoring/debugging
 	if snh.manager != nil {
 		connID := conn.GetID()
-
 		// Track the operation (ignore errors since this is optional)
 		_ = snh.manager.TrackMovingOperationWithConnID(context.Background(), newEndpoint, deadline, seqID, connID)
 	} else {
@@ -160,14 +162,18 @@ func (snh *NotificationHandler) handleMigrating(ctx context.Context, handlerCtx
 		return ErrInvalidNotification
 	}
 
-	connAdapter, ok := handlerCtx.Conn.(interfaces.ConnectionWithRelaxedTimeout)
+	conn, ok := handlerCtx.Conn.(*pool.Conn)
 	if !ok {
 		internal.Logger.Printf(ctx, "hitless: invalid connection type in handler context for MIGRATING notification")
 		return ErrInvalidNotification
 	}
 
 	// Apply relaxed timeout to this specific connection
-	connAdapter.SetRelaxedTimeout(snh.manager.config.RelaxedTimeout, snh.manager.config.RelaxedTimeout)
+	if snh.manager.config.LogLevel >= 3 { // Debug level
+		internal.Logger.Printf(ctx, "hitless: conn[%d] applying relaxed timeout (%v) for MIGRATING notification",
+			snh.manager.config.RelaxedTimeout, conn.GetID())
+	}
+	conn.SetRelaxedTimeout(snh.manager.config.RelaxedTimeout, snh.manager.config.RelaxedTimeout)
 	return nil
 }
 
@@ -185,14 +191,18 @@ func (snh *NotificationHandler) handleMigrated(ctx context.Context, handlerCtx p
 		return ErrInvalidNotification
 	}
 
-	connAdapter, ok := handlerCtx.Conn.(interfaces.ConnectionWithRelaxedTimeout)
+	conn, ok := handlerCtx.Conn.(*pool.Conn)
 	if !ok {
 		internal.Logger.Printf(ctx, "hitless: invalid connection type in handler context for MIGRATED notification")
 		return ErrInvalidNotification
 	}
 
 	// Clear relaxed timeout for this specific connection
-	connAdapter.ClearRelaxedTimeout()
+	if snh.manager.config.LogLevel >= 3 { // Debug level
+		connID := conn.GetID()
+		internal.Logger.Printf(ctx, "hitless: conn[%d] clearing relaxed timeout for MIGRATED notification", connID)
+	}
+	conn.ClearRelaxedTimeout()
 	return nil
 }
 
@@ -210,14 +220,18 @@ func (snh *NotificationHandler) handleFailingOver(ctx context.Context, handlerCt
 		return ErrInvalidNotification
 	}
 
-	connAdapter, ok := handlerCtx.Conn.(interfaces.ConnectionWithRelaxedTimeout)
+	conn, ok := handlerCtx.Conn.(*pool.Conn)
 	if !ok {
 		internal.Logger.Printf(ctx, "hitless: invalid connection type in handler context for FAILING_OVER notification")
 		return ErrInvalidNotification
 	}
 
 	// Apply relaxed timeout to this specific connection
-	connAdapter.SetRelaxedTimeout(snh.manager.config.RelaxedTimeout, snh.manager.config.RelaxedTimeout)
+	if snh.manager.config.LogLevel >= 3 { // Debug level
+		connID := conn.GetID()
+		internal.Logger.Printf(ctx, "hitless: conn[%d] applying relaxed timeout (%v) for FAILING_OVER notification", connID, snh.manager.config.RelaxedTimeout)
+	}
+	conn.SetRelaxedTimeout(snh.manager.config.RelaxedTimeout, snh.manager.config.RelaxedTimeout)
 	return nil
 }
 
@@ -235,13 +249,17 @@ func (snh *NotificationHandler) handleFailedOver(ctx context.Context, handlerCtx
 		return ErrInvalidNotification
 	}
 
-	connAdapter, ok := handlerCtx.Conn.(interfaces.ConnectionWithRelaxedTimeout)
+	conn, ok := handlerCtx.Conn.(*pool.Conn)
 	if !ok {
 		internal.Logger.Printf(ctx, "hitless: invalid connection type in handler context for FAILED_OVER notification")
 		return ErrInvalidNotification
 	}
 
 	// Clear relaxed timeout for this specific connection
-	connAdapter.ClearRelaxedTimeout()
+	if snh.manager.config.LogLevel >= 3 { // Debug level
+		connID := conn.GetID()
+		internal.Logger.Printf(ctx, "hitless: conn[%d] clearing relaxed timeout for FAILED_OVER notification", connID)
+	}
+	conn.ClearRelaxedTimeout()
 	return nil
 }

hitless/pool_hook.go

@@ -68,8 +68,10 @@ func NewPoolHook(baseDialer func(context.Context, string, string) (net.Conn, err
 
 // NewPoolHookWithPoolSize creates a new pool hook with pool size for better worker defaults
 func NewPoolHookWithPoolSize(baseDialer func(context.Context, string, string) (net.Conn, error), network string, config *Config, hitlessManager HitlessManagerInterface, poolSize int) *PoolHook {
-	// Apply defaults to any missing configuration fields, using pool size for worker calculations
-	config = config.ApplyDefaultsWithPoolSize(poolSize)
+	// Apply defaults if config is nil or has zero values
+	if config == nil {
+		config = config.ApplyDefaultsWithPoolSize(poolSize)
+	}
 
 	ph := &PoolHook{
 		// baseDialer is used to create connections to new endpoints during handoffs
@@ -86,9 +88,6 @@ func NewPoolHookWithPoolSize(baseDialer func(context.Context, string, string) (n
 		// Hitless manager for operation completion tracking
 		hitlessManager: hitlessManager,
 	}
-
-	// No upfront worker creation - workers are created on demand
-
 	return ph
 }
 

hitless/pool_hook_test.go

@@ -528,6 +528,7 @@ func TestConnectionHook(t *testing.T) {
 		config := &Config{
 			MaxWorkers:                 2,
 			HandoffQueueSize:           10,
+			MaxHandoffRetries:          3, // Allow retries for successful handoff
 			PostHandoffRelaxedDuration: 100 * time.Millisecond, // Fast expiration for testing
 			RelaxedTimeout:             5 * time.Second,
 			LogLevel:                   2,
@@ -805,6 +806,7 @@ func TestConnectionHook(t *testing.T) {
 		config := &Config{
 			MaxWorkers:                 2,
 			HandoffQueueSize:           10,
+			MaxHandoffRetries:          3, // Allow retries for successful handoff
 			RelaxedTimeout:             5 * time.Second,
 			PostHandoffRelaxedDuration: 100 * time.Millisecond, // Short for testing
 		}

internal/interfaces/interfaces.go

@@ -8,7 +8,8 @@ import (
 	"time"
 )
 
-// Forward declaration to avoid circular imports
+// NotificationProcessor is (most probably) a push.NotificationProcessor
+// forward declaration to avoid circular imports
 type NotificationProcessor interface {
 	RegisterHandler(pushNotificationName string, handler interface{}, protected bool) error
 	UnregisterHandler(pushNotificationName string) error
@@ -25,6 +26,7 @@ type ClientInterface interface {
 }
 
 // OptionsInterface defines the interface for client options.
+// Uses an adapter pattern to avoid circular dependencies.
 type OptionsInterface interface {
 	// GetReadTimeout returns the read timeout.
 	GetReadTimeout() time.Duration
@@ -50,18 +52,3 @@ type OptionsInterface interface {
 	// NewDialer returns a new dialer function for the connection.
 	NewDialer() func(context.Context) (net.Conn, error)
 }
-
-// ConnectionWithRelaxedTimeout defines the interface for connections that support relaxed timeout adjustment.
-// This is used by the hitless upgrade system for per-connection timeout management.
-type ConnectionWithRelaxedTimeout interface {
-	// SetRelaxedTimeout sets relaxed timeouts for this connection during hitless upgrades.
-	// These timeouts remain active until explicitly cleared.
-	SetRelaxedTimeout(readTimeout, writeTimeout time.Duration)
-
-	// SetRelaxedTimeoutWithDeadline sets relaxed timeouts with an expiration deadline.
-	// After the deadline, timeouts automatically revert to normal values.
-	SetRelaxedTimeoutWithDeadline(readTimeout, writeTimeout time.Duration, deadline time.Time)
-
-	// ClearRelaxedTimeout clears relaxed timeouts for this connection.
-	ClearRelaxedTimeout()
-}

pubsub.go

@@ -485,11 +485,10 @@ func (c *PubSub) ReceiveTimeout(ctx context.Context, timeout time.Duration) (int
 		if err := c.processPendingPushNotificationWithReader(ctx, cn, rd); err != nil {
 			// Log the error but don't fail the command execution
 			// Push notification processing errors shouldn't break normal Redis operations
-			internal.Logger.Printf(ctx, "push: error processing pending notifications before reading reply: %v", err)
+			internal.Logger.Printf(ctx, "push: conn[%d] error processing pending notifications before reading reply: %v", cn.GetID(), err)
 		}
 		return c.cmd.readReply(rd)
 	})
-
 	c.releaseConnWithLock(ctx, cn, err, timeout > 0)
 
 	if err != nil {