mirror of
https://github.com/redis/go-redis.git
synced 2025-09-07 07:47:24 +03:00
fix hooks and add logging, logging will be removed before merge
This commit is contained in:
@@ -103,8 +103,8 @@ func (ca *connectionAdapter) IsUsable() bool {
|
|||||||
return ca.conn.IsUsable()
|
return ca.conn.IsUsable()
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetPoolConnection returns the underlying pool connection.
|
// GetPoolConn returns the underlying pool connection.
|
||||||
func (ca *connectionAdapter) GetPoolConnection() *pool.Conn {
|
func (ca *connectionAdapter) GetPoolConn() *pool.Conn {
|
||||||
return ca.conn
|
return ca.conn
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@@ -3,6 +3,7 @@ package hitless
|
|||||||
import (
|
import (
|
||||||
"net"
|
"net"
|
||||||
"runtime"
|
"runtime"
|
||||||
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/redis/go-redis/v9/internal/util"
|
"github.com/redis/go-redis/v9/internal/util"
|
||||||
@@ -183,8 +184,6 @@ func (c *Config) Validate() error {
|
|||||||
return ErrInvalidHandoffRetries
|
return ErrInvalidHandoffRetries
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -284,8 +283,6 @@ func (c *Config) ApplyDefaultsWithPoolSize(poolSize int) *Config {
|
|||||||
result.MaxHandoffRetries = c.MaxHandoffRetries
|
result.MaxHandoffRetries = c.MaxHandoffRetries
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
return result
|
return result
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -334,44 +331,92 @@ func (c *Config) applyWorkerDefaults(poolSize int) {
|
|||||||
|
|
||||||
// DetectEndpointType automatically detects the appropriate endpoint type
|
// DetectEndpointType automatically detects the appropriate endpoint type
|
||||||
// based on the connection address and TLS configuration.
|
// based on the connection address and TLS configuration.
|
||||||
|
//
|
||||||
|
// For IP addresses:
|
||||||
|
// - If TLS is enabled: requests FQDN for proper certificate validation
|
||||||
|
// - If TLS is disabled: requests IP for better performance
|
||||||
|
//
|
||||||
|
// For hostnames:
|
||||||
|
// - If TLS is enabled: always requests FQDN for proper certificate validation
|
||||||
|
// - If TLS is disabled: requests IP for better performance
|
||||||
|
//
|
||||||
|
// Internal vs External detection:
|
||||||
|
// - For IPs: uses private IP range detection
|
||||||
|
// - For hostnames: uses heuristics based on common internal naming patterns
|
||||||
func DetectEndpointType(addr string, tlsEnabled bool) EndpointType {
|
func DetectEndpointType(addr string, tlsEnabled bool) EndpointType {
|
||||||
// Parse the address to determine if it's an IP or hostname
|
|
||||||
isPrivate := isPrivateIP(addr)
|
|
||||||
|
|
||||||
var endpointType EndpointType
|
|
||||||
|
|
||||||
if tlsEnabled {
|
|
||||||
// TLS requires FQDN for certificate validation
|
|
||||||
if isPrivate {
|
|
||||||
endpointType = EndpointTypeInternalFQDN
|
|
||||||
} else {
|
|
||||||
endpointType = EndpointTypeExternalFQDN
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
// No TLS, can use IP addresses
|
|
||||||
if isPrivate {
|
|
||||||
endpointType = EndpointTypeInternalIP
|
|
||||||
} else {
|
|
||||||
endpointType = EndpointTypeExternalIP
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return endpointType
|
|
||||||
}
|
|
||||||
|
|
||||||
// isPrivateIP checks if the given address is in a private IP range.
|
|
||||||
func isPrivateIP(addr string) bool {
|
|
||||||
// Extract host from "host:port" format
|
// Extract host from "host:port" format
|
||||||
host, _, err := net.SplitHostPort(addr)
|
host, _, err := net.SplitHostPort(addr)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
host = addr // Assume no port
|
host = addr // Assume no port
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Check if the host is an IP address or hostname
|
||||||
ip := net.ParseIP(host)
|
ip := net.ParseIP(host)
|
||||||
if ip == nil {
|
isIPAddress := ip != nil
|
||||||
return false // Not an IP address (likely hostname)
|
var endpointType EndpointType
|
||||||
|
|
||||||
|
if isIPAddress {
|
||||||
|
// Address is an IP - determine if it's private or public
|
||||||
|
isPrivate := ip.IsPrivate() || ip.IsLoopback() || ip.IsLinkLocalUnicast()
|
||||||
|
|
||||||
|
if tlsEnabled {
|
||||||
|
// TLS with IP addresses - still prefer FQDN for certificate validation
|
||||||
|
if isPrivate {
|
||||||
|
endpointType = EndpointTypeInternalFQDN
|
||||||
|
} else {
|
||||||
|
endpointType = EndpointTypeExternalFQDN
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
// No TLS - can use IP addresses directly
|
||||||
|
if isPrivate {
|
||||||
|
endpointType = EndpointTypeInternalIP
|
||||||
|
} else {
|
||||||
|
endpointType = EndpointTypeExternalIP
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
// Address is a hostname
|
||||||
|
isInternalHostname := isInternalHostname(host)
|
||||||
|
if isInternalHostname {
|
||||||
|
endpointType = EndpointTypeInternalFQDN
|
||||||
|
} else {
|
||||||
|
endpointType = EndpointTypeExternalFQDN
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Check for private/loopback ranges
|
return endpointType
|
||||||
return ip.IsPrivate() || ip.IsLoopback() || ip.IsLinkLocalUnicast()
|
}
|
||||||
|
|
||||||
|
// isInternalHostname determines if a hostname appears to be internal/private.
|
||||||
|
// This is a heuristic based on common naming patterns.
|
||||||
|
func isInternalHostname(hostname string) bool {
|
||||||
|
// Convert to lowercase for comparison
|
||||||
|
hostname = strings.ToLower(hostname)
|
||||||
|
|
||||||
|
// Common internal hostname patterns
|
||||||
|
internalPatterns := []string{
|
||||||
|
"localhost",
|
||||||
|
".local",
|
||||||
|
".internal",
|
||||||
|
".corp",
|
||||||
|
".lan",
|
||||||
|
".intranet",
|
||||||
|
".private",
|
||||||
|
}
|
||||||
|
|
||||||
|
// Check for exact match or suffix match
|
||||||
|
for _, pattern := range internalPatterns {
|
||||||
|
if hostname == pattern || strings.HasSuffix(hostname, pattern) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Check for RFC 1918 style hostnames (e.g., redis-1, db-server, etc.)
|
||||||
|
// If hostname doesn't contain dots, it's likely internal
|
||||||
|
if !strings.Contains(hostname, ".") {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
// Default to external for fully qualified domain names
|
||||||
|
return false
|
||||||
}
|
}
|
||||||
|
@@ -13,8 +13,6 @@ import (
|
|||||||
"github.com/redis/go-redis/v9/internal/pool"
|
"github.com/redis/go-redis/v9/internal/pool"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
// Push notification type constants for hitless upgrades
|
// Push notification type constants for hitless upgrades
|
||||||
const (
|
const (
|
||||||
NotificationMoving = "MOVING"
|
NotificationMoving = "MOVING"
|
||||||
@@ -297,3 +295,9 @@ func (hm *HitlessManager) createPoolHook(baseDialer func(context.Context, string
|
|||||||
|
|
||||||
return hm.poolHooksRef
|
return hm.poolHooksRef
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (hm *HitlessManager) AddNotificationHook(notificationHook NotificationHook) {
|
||||||
|
hm.hooksMu.Lock()
|
||||||
|
defer hm.hooksMu.Unlock()
|
||||||
|
hm.hooks = append(hm.hooks, notificationHook)
|
||||||
|
}
|
||||||
|
@@ -22,27 +22,14 @@ func (lh *LoggingHook) PreHook(ctx context.Context, notificationType string, not
|
|||||||
// PostHook logs the result after processing.
|
// PostHook logs the result after processing.
|
||||||
func (lh *LoggingHook) PostHook(ctx context.Context, notificationType string, notification []interface{}, result error) {
|
func (lh *LoggingHook) PostHook(ctx context.Context, notificationType string, notification []interface{}, result error) {
|
||||||
if result != nil && lh.LogLevel >= 1 { // Warning level
|
if result != nil && lh.LogLevel >= 1 { // Warning level
|
||||||
internal.Logger.Printf(ctx, "hitless: %s notification processing failed: %v", notificationType, result)
|
internal.Logger.Printf(ctx, "hitless: %s notification processing failed: %v - %v", notificationType, result, notification)
|
||||||
} else if lh.LogLevel >= 3 { // Debug level
|
} else if lh.LogLevel >= 3 { // Debug level
|
||||||
internal.Logger.Printf(ctx, "hitless: %s notification processed successfully", notificationType)
|
internal.Logger.Printf(ctx, "hitless: %s notification processed successfully", notificationType)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// FilterHook is an example hook that can filter out certain notifications.
|
// NewLoggingHook creates a new logging hook with the specified log level.
|
||||||
type FilterHook struct {
|
// Log levels: 0=errors, 1=warnings, 2=info, 3=debug
|
||||||
BlockedTypes map[string]bool
|
func NewLoggingHook(logLevel int) *LoggingHook {
|
||||||
}
|
return &LoggingHook{LogLevel: logLevel}
|
||||||
|
|
||||||
// PreHook filters notifications based on type.
|
|
||||||
func (fh *FilterHook) PreHook(ctx context.Context, notificationType string, notification []interface{}) ([]interface{}, bool) {
|
|
||||||
if fh.BlockedTypes[notificationType] {
|
|
||||||
internal.Logger.Printf(ctx, "hitless: filtering out %s notification", notificationType)
|
|
||||||
return notification, false // Skip processing
|
|
||||||
}
|
|
||||||
return notification, true
|
|
||||||
}
|
|
||||||
|
|
||||||
// PostHook does nothing for filter hook.
|
|
||||||
func (fh *FilterHook) PostHook(ctx context.Context, notificationType string, notification []interface{}, result error) {
|
|
||||||
// No post-processing needed for filter hook
|
|
||||||
}
|
}
|
||||||
|
@@ -19,11 +19,13 @@ type NotificationHandler struct {
|
|||||||
// HandlePushNotification processes push notifications with hook support.
|
// HandlePushNotification processes push notifications with hook support.
|
||||||
func (snh *NotificationHandler) HandlePushNotification(ctx context.Context, handlerCtx push.NotificationHandlerContext, notification []interface{}) error {
|
func (snh *NotificationHandler) HandlePushNotification(ctx context.Context, handlerCtx push.NotificationHandlerContext, notification []interface{}) error {
|
||||||
if len(notification) == 0 {
|
if len(notification) == 0 {
|
||||||
|
internal.Logger.Printf(ctx, "hitless: invalid notification format: %v", notification)
|
||||||
return ErrInvalidNotification
|
return ErrInvalidNotification
|
||||||
}
|
}
|
||||||
|
|
||||||
notificationType, ok := notification[0].(string)
|
notificationType, ok := notification[0].(string)
|
||||||
if !ok {
|
if !ok {
|
||||||
|
internal.Logger.Printf(ctx, "hitless: invalid notification type format: %v", notification[0])
|
||||||
return ErrInvalidNotification
|
return ErrInvalidNotification
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -60,16 +62,19 @@ func (snh *NotificationHandler) HandlePushNotification(ctx context.Context, hand
|
|||||||
// ["MOVING", seqNum, timeS, endpoint] - per-connection handoff
|
// ["MOVING", seqNum, timeS, endpoint] - per-connection handoff
|
||||||
func (snh *NotificationHandler) handleMoving(ctx context.Context, handlerCtx push.NotificationHandlerContext, notification []interface{}) error {
|
func (snh *NotificationHandler) handleMoving(ctx context.Context, handlerCtx push.NotificationHandlerContext, notification []interface{}) error {
|
||||||
if len(notification) < 3 {
|
if len(notification) < 3 {
|
||||||
|
internal.Logger.Printf(ctx, "hitless: invalid MOVING notification: %v", notification)
|
||||||
return ErrInvalidNotification
|
return ErrInvalidNotification
|
||||||
}
|
}
|
||||||
seqID, ok := notification[1].(int64)
|
seqID, ok := notification[1].(int64)
|
||||||
if !ok {
|
if !ok {
|
||||||
|
internal.Logger.Printf(ctx, "hitless: invalid seqID in MOVING notification: %v", notification[1])
|
||||||
return ErrInvalidNotification
|
return ErrInvalidNotification
|
||||||
}
|
}
|
||||||
|
|
||||||
// Extract timeS
|
// Extract timeS
|
||||||
timeS, ok := notification[2].(int64)
|
timeS, ok := notification[2].(int64)
|
||||||
if !ok {
|
if !ok {
|
||||||
|
internal.Logger.Printf(ctx, "hitless: invalid timeS in MOVING notification: %v", notification[2])
|
||||||
return ErrInvalidNotification
|
return ErrInvalidNotification
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -78,6 +83,7 @@ func (snh *NotificationHandler) handleMoving(ctx context.Context, handlerCtx pus
|
|||||||
// Extract new endpoint
|
// Extract new endpoint
|
||||||
newEndpoint, ok = notification[3].(string)
|
newEndpoint, ok = notification[3].(string)
|
||||||
if !ok {
|
if !ok {
|
||||||
|
internal.Logger.Printf(ctx, "hitless: invalid newEndpoint in MOVING notification: %v", notification[3])
|
||||||
return ErrInvalidNotification
|
return ErrInvalidNotification
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -85,6 +91,7 @@ func (snh *NotificationHandler) handleMoving(ctx context.Context, handlerCtx pus
|
|||||||
// Get the connection that received this notification
|
// Get the connection that received this notification
|
||||||
conn := handlerCtx.Conn
|
conn := handlerCtx.Conn
|
||||||
if conn == nil {
|
if conn == nil {
|
||||||
|
internal.Logger.Printf(ctx, "hitless: no connection in handler context for MOVING notification")
|
||||||
return ErrInvalidNotification
|
return ErrInvalidNotification
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -95,6 +102,7 @@ func (snh *NotificationHandler) handleMoving(ctx context.Context, handlerCtx pus
|
|||||||
} else if pc, ok := conn.(*pool.Conn); ok {
|
} else if pc, ok := conn.(*pool.Conn); ok {
|
||||||
poolConn = pc
|
poolConn = pc
|
||||||
} else {
|
} else {
|
||||||
|
internal.Logger.Printf(ctx, "hitless: invalid connection type in handler context for MOVING notification - %T %#v", conn, handlerCtx)
|
||||||
return ErrInvalidNotification
|
return ErrInvalidNotification
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -145,17 +153,20 @@ func (snh *NotificationHandler) handleMigrating(ctx context.Context, handlerCtx
|
|||||||
// MIGRATING notifications indicate that a connection is about to be migrated
|
// MIGRATING notifications indicate that a connection is about to be migrated
|
||||||
// Apply relaxed timeouts to the specific connection that received this notification
|
// Apply relaxed timeouts to the specific connection that received this notification
|
||||||
if len(notification) < 2 {
|
if len(notification) < 2 {
|
||||||
|
internal.Logger.Printf(ctx, "hitless: invalid MIGRATING notification: %v", notification)
|
||||||
return ErrInvalidNotification
|
return ErrInvalidNotification
|
||||||
}
|
}
|
||||||
|
|
||||||
// Get the connection from handler context and type assert to connectionAdapter
|
// Get the connection from handler context and type assert to connectionAdapter
|
||||||
if handlerCtx.Conn == nil {
|
if handlerCtx.Conn == nil {
|
||||||
|
internal.Logger.Printf(ctx, "hitless: no connection in handler context for MIGRATING notification")
|
||||||
return ErrInvalidNotification
|
return ErrInvalidNotification
|
||||||
}
|
}
|
||||||
|
|
||||||
// Type assert to connectionAdapter which implements ConnectionWithRelaxedTimeout
|
// Type assert to connectionAdapter which implements ConnectionWithRelaxedTimeout
|
||||||
connAdapter, ok := handlerCtx.Conn.(interfaces.ConnectionWithRelaxedTimeout)
|
connAdapter, ok := handlerCtx.Conn.(interfaces.ConnectionWithRelaxedTimeout)
|
||||||
if !ok {
|
if !ok {
|
||||||
|
internal.Logger.Printf(ctx, "hitless: invalid connection type in handler context for MIGRATING notification")
|
||||||
return ErrInvalidNotification
|
return ErrInvalidNotification
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -169,17 +180,20 @@ func (snh *NotificationHandler) handleMigrated(ctx context.Context, handlerCtx p
|
|||||||
// MIGRATED notifications indicate that a connection migration has completed
|
// MIGRATED notifications indicate that a connection migration has completed
|
||||||
// Restore normal timeouts for the specific connection that received this notification
|
// Restore normal timeouts for the specific connection that received this notification
|
||||||
if len(notification) < 2 {
|
if len(notification) < 2 {
|
||||||
|
internal.Logger.Printf(ctx, "hitless: invalid MIGRATED notification: %v", notification)
|
||||||
return ErrInvalidNotification
|
return ErrInvalidNotification
|
||||||
}
|
}
|
||||||
|
|
||||||
// Get the connection from handler context and type assert to connectionAdapter
|
// Get the connection from handler context and type assert to connectionAdapter
|
||||||
if handlerCtx.Conn == nil {
|
if handlerCtx.Conn == nil {
|
||||||
|
internal.Logger.Printf(ctx, "hitless: no connection in handler context for MIGRATED notification")
|
||||||
return ErrInvalidNotification
|
return ErrInvalidNotification
|
||||||
}
|
}
|
||||||
|
|
||||||
// Type assert to connectionAdapter which implements ConnectionWithRelaxedTimeout
|
// Type assert to connectionAdapter which implements ConnectionWithRelaxedTimeout
|
||||||
connAdapter, ok := handlerCtx.Conn.(interfaces.ConnectionWithRelaxedTimeout)
|
connAdapter, ok := handlerCtx.Conn.(interfaces.ConnectionWithRelaxedTimeout)
|
||||||
if !ok {
|
if !ok {
|
||||||
|
internal.Logger.Printf(ctx, "hitless: invalid connection type in handler context for MIGRATED notification")
|
||||||
return ErrInvalidNotification
|
return ErrInvalidNotification
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -193,17 +207,20 @@ func (snh *NotificationHandler) handleFailingOver(ctx context.Context, handlerCt
|
|||||||
// FAILING_OVER notifications indicate that a connection is about to failover
|
// FAILING_OVER notifications indicate that a connection is about to failover
|
||||||
// Apply relaxed timeouts to the specific connection that received this notification
|
// Apply relaxed timeouts to the specific connection that received this notification
|
||||||
if len(notification) < 2 {
|
if len(notification) < 2 {
|
||||||
|
internal.Logger.Printf(ctx, "hitless: invalid FAILING_OVER notification: %v", notification)
|
||||||
return ErrInvalidNotification
|
return ErrInvalidNotification
|
||||||
}
|
}
|
||||||
|
|
||||||
// Get the connection from handler context and type assert to connectionAdapter
|
// Get the connection from handler context and type assert to connectionAdapter
|
||||||
if handlerCtx.Conn == nil {
|
if handlerCtx.Conn == nil {
|
||||||
|
internal.Logger.Printf(ctx, "hitless: no connection in handler context for FAILING_OVER notification")
|
||||||
return ErrInvalidNotification
|
return ErrInvalidNotification
|
||||||
}
|
}
|
||||||
|
|
||||||
// Type assert to connectionAdapter which implements ConnectionWithRelaxedTimeout
|
// Type assert to connectionAdapter which implements ConnectionWithRelaxedTimeout
|
||||||
connAdapter, ok := handlerCtx.Conn.(interfaces.ConnectionWithRelaxedTimeout)
|
connAdapter, ok := handlerCtx.Conn.(interfaces.ConnectionWithRelaxedTimeout)
|
||||||
if !ok {
|
if !ok {
|
||||||
|
internal.Logger.Printf(ctx, "hitless: invalid connection type in handler context for FAILING_OVER notification")
|
||||||
return ErrInvalidNotification
|
return ErrInvalidNotification
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -217,17 +234,20 @@ func (snh *NotificationHandler) handleFailedOver(ctx context.Context, handlerCtx
|
|||||||
// FAILED_OVER notifications indicate that a connection failover has completed
|
// FAILED_OVER notifications indicate that a connection failover has completed
|
||||||
// Restore normal timeouts for the specific connection that received this notification
|
// Restore normal timeouts for the specific connection that received this notification
|
||||||
if len(notification) < 2 {
|
if len(notification) < 2 {
|
||||||
|
internal.Logger.Printf(ctx, "hitless: invalid FAILED_OVER notification: %v", notification)
|
||||||
return ErrInvalidNotification
|
return ErrInvalidNotification
|
||||||
}
|
}
|
||||||
|
|
||||||
// Get the connection from handler context and type assert to connectionAdapter
|
// Get the connection from handler context and type assert to connectionAdapter
|
||||||
if handlerCtx.Conn == nil {
|
if handlerCtx.Conn == nil {
|
||||||
|
internal.Logger.Printf(ctx, "hitless: no connection in handler context for FAILED_OVER notification")
|
||||||
return ErrInvalidNotification
|
return ErrInvalidNotification
|
||||||
}
|
}
|
||||||
|
|
||||||
// Type assert to connectionAdapter which implements ConnectionWithRelaxedTimeout
|
// Type assert to connectionAdapter which implements ConnectionWithRelaxedTimeout
|
||||||
connAdapter, ok := handlerCtx.Conn.(interfaces.ConnectionWithRelaxedTimeout)
|
connAdapter, ok := handlerCtx.Conn.(interfaces.ConnectionWithRelaxedTimeout)
|
||||||
if !ok {
|
if !ok {
|
||||||
|
internal.Logger.Printf(ctx, "hitless: invalid connection type in handler context for FAILED_OVER notification")
|
||||||
return ErrInvalidNotification
|
return ErrInvalidNotification
|
||||||
}
|
}
|
||||||
|
|
||||||
|
Reference in New Issue
Block a user