Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import enum
2import ipaddress
3import logging
4import re
5import threading
6import time
7from abc import ABC, abstractmethod
8from typing import (
9 TYPE_CHECKING,
10 Any,
11 Dict,
12 List,
13 Literal,
14 Mapping,
15 Optional,
16 Type,
17 Union,
18)
20from redis.event import (
21 EventDispatcherInterface,
22 MaintenanceCompletedEvent,
23 MaintenanceEvent,
24 MaintenanceStartedEvent,
25)
26from redis.observability.attributes import get_pool_name
27from redis.observability.recorder import (
28 record_connection_handoff,
29 record_connection_relaxed_timeout,
30 record_maint_notification_count,
31)
32from redis.typing import Number
34if TYPE_CHECKING:
35 from redis.cluster import MaintNotificationsAbstractRedisCluster
37logger = logging.getLogger(__name__)
40class MaintenanceState(enum.Enum):
41 NONE = "none"
42 MOVING = "moving"
43 MAINTENANCE = "maintenance"
46class EndpointType(enum.Enum):
47 """Valid endpoint types used in CLIENT MAINT_NOTIFICATIONS command."""
49 INTERNAL_IP = "internal-ip"
50 INTERNAL_FQDN = "internal-fqdn"
51 EXTERNAL_IP = "external-ip"
52 EXTERNAL_FQDN = "external-fqdn"
53 NONE = "none"
55 def __str__(self):
56 """Return the string value of the enum."""
57 return self.value
60if TYPE_CHECKING:
61 from redis.asyncio.connection import AsyncMaintNotificationsAbstractConnection
62 from redis.connection import (
63 MaintNotificationsAbstractConnection,
64 MaintNotificationsAbstractConnectionPool,
65 )
68class MaintenanceNotification(ABC):
69 """
70 Base class for maintenance notifications sent through push messages by Redis server.
72 This class provides common functionality for all maintenance notifications including
73 unique identification and TTL (Time-To-Live) functionality.
75 Attributes:
76 id (int): Unique identifier for this notification
77 ttl (int): Time-to-live in seconds for this notification
78 creation_time (float): Timestamp when the notification was created/read
79 """
81 def __init__(self, id: int, ttl: int):
82 """
83 Initialize a new MaintenanceNotification with unique ID and TTL functionality.
85 Args:
86 id (int): Unique identifier for this notification
87 ttl (int): Time-to-live in seconds for this notification
88 """
89 self.id = id
90 self.ttl = ttl
91 self.creation_time = time.monotonic()
92 self.expire_at = self.creation_time + self.ttl
94 def is_expired(self) -> bool:
95 """
96 Check if this notification has expired based on its TTL
97 and creation time.
99 Returns:
100 bool: True if the notification has expired, False otherwise
101 """
102 return time.monotonic() > (self.creation_time + self.ttl)
104 @abstractmethod
105 def __repr__(self) -> str:
106 """
107 Return a string representation of the maintenance notification.
109 This method must be implemented by all concrete subclasses.
111 Returns:
112 str: String representation of the notification
113 """
114 pass
116 @abstractmethod
117 def __eq__(self, other) -> bool:
118 """
119 Compare two maintenance notifications for equality.
121 This method must be implemented by all concrete subclasses.
122 Notifications are typically considered equal if they have the same id
123 and are of the same type.
125 Args:
126 other: The other object to compare with
128 Returns:
129 bool: True if the notifications are equal, False otherwise
130 """
131 pass
133 @abstractmethod
134 def __hash__(self) -> int:
135 """
136 Return a hash value for the maintenance notification.
138 This method must be implemented by all concrete subclasses to allow
139 instances to be used in sets and as dictionary keys.
141 Returns:
142 int: Hash value for the notification
143 """
144 pass
147class NodeMovingNotification(MaintenanceNotification):
148 """
149 This notification is received when a node is replaced with a new node
150 during cluster rebalancing or maintenance operations.
151 """
153 def __init__(
154 self,
155 id: int,
156 new_node_host: Optional[str],
157 new_node_port: Optional[int],
158 ttl: int,
159 ):
160 """
161 Initialize a new NodeMovingNotification.
163 Args:
164 id (int): Unique identifier for this notification
165 new_node_host (str): Hostname or IP address of the new replacement node
166 new_node_port (int): Port number of the new replacement node
167 ttl (int): Time-to-live in seconds for this notification
168 """
169 super().__init__(id, ttl)
170 self.new_node_host = new_node_host
171 self.new_node_port = new_node_port
173 def __repr__(self) -> str:
174 expiry_time = self.expire_at
175 remaining = max(0, expiry_time - time.monotonic())
177 return (
178 f"{self.__class__.__name__}("
179 f"id={self.id}, "
180 f"new_node_host='{self.new_node_host}', "
181 f"new_node_port={self.new_node_port}, "
182 f"ttl={self.ttl}, "
183 f"creation_time={self.creation_time}, "
184 f"expires_at={expiry_time}, "
185 f"remaining={remaining:.1f}s, "
186 f"expired={self.is_expired()}"
187 f")"
188 )
190 def __eq__(self, other) -> bool:
191 """
192 Two NodeMovingNotification notifications are considered equal if they have the same
193 id, new_node_host, and new_node_port.
194 """
195 if not isinstance(other, NodeMovingNotification):
196 return False
197 return (
198 self.id == other.id
199 and self.new_node_host == other.new_node_host
200 and self.new_node_port == other.new_node_port
201 )
203 def __hash__(self) -> int:
204 """
205 Return a hash value for the notification to allow
206 instances to be used in sets and as dictionary keys.
208 Returns:
209 int: Hash value based on notification type class name, id,
210 new_node_host and new_node_port
211 """
212 try:
213 node_port = int(self.new_node_port) if self.new_node_port else None
214 except ValueError:
215 node_port = 0
217 return hash(
218 (
219 self.__class__.__name__,
220 int(self.id),
221 str(self.new_node_host),
222 node_port,
223 )
224 )
227class NodeMigratingNotification(MaintenanceNotification):
228 """
229 Notification for when a Redis cluster node is in the process of migrating slots.
231 This notification is received when a node starts migrating its slots to another node
232 during cluster rebalancing or maintenance operations.
234 Args:
235 id (int): Unique identifier for this notification
236 ttl (int): Time-to-live in seconds for this notification
237 """
239 def __init__(self, id: int, ttl: int):
240 super().__init__(id, ttl)
242 def __repr__(self) -> str:
243 expiry_time = self.creation_time + self.ttl
244 remaining = max(0, expiry_time - time.monotonic())
245 return (
246 f"{self.__class__.__name__}("
247 f"id={self.id}, "
248 f"ttl={self.ttl}, "
249 f"creation_time={self.creation_time}, "
250 f"expires_at={expiry_time}, "
251 f"remaining={remaining:.1f}s, "
252 f"expired={self.is_expired()}"
253 f")"
254 )
256 def __eq__(self, other) -> bool:
257 """
258 Two NodeMigratingNotification notifications are considered equal if they have the same
259 id and are of the same type.
260 """
261 if not isinstance(other, NodeMigratingNotification):
262 return False
263 return self.id == other.id and type(self) is type(other)
265 def __hash__(self) -> int:
266 """
267 Return a hash value for the notification to allow
268 instances to be used in sets and as dictionary keys.
270 Returns:
271 int: Hash value based on notification type and id
272 """
273 return hash((self.__class__.__name__, int(self.id)))
276class NodeMigratedNotification(MaintenanceNotification):
277 """
278 Notification for when a Redis cluster node has completed migrating slots.
280 This notification is received when a node has finished migrating all its slots
281 to other nodes during cluster rebalancing or maintenance operations.
283 Args:
284 id (int): Unique identifier for this notification
285 """
287 DEFAULT_TTL = 5
289 def __init__(self, id: int):
290 super().__init__(id, NodeMigratedNotification.DEFAULT_TTL)
292 def __repr__(self) -> str:
293 expiry_time = self.creation_time + self.ttl
294 remaining = max(0, expiry_time - time.monotonic())
295 return (
296 f"{self.__class__.__name__}("
297 f"id={self.id}, "
298 f"ttl={self.ttl}, "
299 f"creation_time={self.creation_time}, "
300 f"expires_at={expiry_time}, "
301 f"remaining={remaining:.1f}s, "
302 f"expired={self.is_expired()}"
303 f")"
304 )
306 def __eq__(self, other) -> bool:
307 """
308 Two NodeMigratedNotification notifications are considered equal if they have the same
309 id and are of the same type.
310 """
311 if not isinstance(other, NodeMigratedNotification):
312 return False
313 return self.id == other.id and type(self) is type(other)
315 def __hash__(self) -> int:
316 """
317 Return a hash value for the notification to allow
318 instances to be used in sets and as dictionary keys.
320 Returns:
321 int: Hash value based on notification type and id
322 """
323 return hash((self.__class__.__name__, int(self.id)))
326class NodeFailingOverNotification(MaintenanceNotification):
327 """
328 Notification for when a Redis cluster node is in the process of failing over.
330 This notification is received when a node starts a failover process during
331 cluster maintenance operations or when handling node failures.
333 Args:
334 id (int): Unique identifier for this notification
335 ttl (int): Time-to-live in seconds for this notification
336 """
338 def __init__(self, id: int, ttl: int):
339 super().__init__(id, ttl)
341 def __repr__(self) -> str:
342 expiry_time = self.creation_time + self.ttl
343 remaining = max(0, expiry_time - time.monotonic())
344 return (
345 f"{self.__class__.__name__}("
346 f"id={self.id}, "
347 f"ttl={self.ttl}, "
348 f"creation_time={self.creation_time}, "
349 f"expires_at={expiry_time}, "
350 f"remaining={remaining:.1f}s, "
351 f"expired={self.is_expired()}"
352 f")"
353 )
355 def __eq__(self, other) -> bool:
356 """
357 Two NodeFailingOverNotification notifications are considered equal if they have the same
358 id and are of the same type.
359 """
360 if not isinstance(other, NodeFailingOverNotification):
361 return False
362 return self.id == other.id and type(self) is type(other)
364 def __hash__(self) -> int:
365 """
366 Return a hash value for the notification to allow
367 instances to be used in sets and as dictionary keys.
369 Returns:
370 int: Hash value based on notification type and id
371 """
372 return hash((self.__class__.__name__, int(self.id)))
375class NodeFailedOverNotification(MaintenanceNotification):
376 """
377 Notification for when a Redis cluster node has completed a failover.
379 This notification is received when a node has finished the failover process
380 during cluster maintenance operations or after handling node failures.
382 Args:
383 id (int): Unique identifier for this notification
384 """
386 DEFAULT_TTL = 5
388 def __init__(self, id: int):
389 super().__init__(id, NodeFailedOverNotification.DEFAULT_TTL)
391 def __repr__(self) -> str:
392 expiry_time = self.creation_time + self.ttl
393 remaining = max(0, expiry_time - time.monotonic())
394 return (
395 f"{self.__class__.__name__}("
396 f"id={self.id}, "
397 f"ttl={self.ttl}, "
398 f"creation_time={self.creation_time}, "
399 f"expires_at={expiry_time}, "
400 f"remaining={remaining:.1f}s, "
401 f"expired={self.is_expired()}"
402 f")"
403 )
405 def __eq__(self, other) -> bool:
406 """
407 Two NodeFailedOverNotification notifications are considered equal if they have the same
408 id and are of the same type.
409 """
410 if not isinstance(other, NodeFailedOverNotification):
411 return False
412 return self.id == other.id and type(self) is type(other)
414 def __hash__(self) -> int:
415 """
416 Return a hash value for the notification to allow
417 instances to be used in sets and as dictionary keys.
419 Returns:
420 int: Hash value based on notification type and id
421 """
422 return hash((self.__class__.__name__, int(self.id)))
425class OSSNodeMigratingNotification(MaintenanceNotification):
426 """
427 Notification for when a Redis OSS API client is used and a node is in the process of migrating slots.
429 This notification is received when a node starts migrating its slots to another node
430 during cluster rebalancing or maintenance operations.
432 Args:
433 id (int): Unique identifier for this notification
434 slots (Optional[List[int]]): List of slots being migrated
435 """
437 DEFAULT_TTL = 30
439 def __init__(
440 self,
441 id: int,
442 slots: Optional[str] = None,
443 ):
444 super().__init__(id, OSSNodeMigratingNotification.DEFAULT_TTL)
445 self.slots = slots
447 def __repr__(self) -> str:
448 expiry_time = self.creation_time + self.ttl
449 remaining = max(0, expiry_time - time.monotonic())
450 return (
451 f"{self.__class__.__name__}("
452 f"id={self.id}, "
453 f"slots={self.slots}, "
454 f"ttl={self.ttl}, "
455 f"creation_time={self.creation_time}, "
456 f"expires_at={expiry_time}, "
457 f"remaining={remaining:.1f}s, "
458 f"expired={self.is_expired()}"
459 f")"
460 )
462 def __eq__(self, other) -> bool:
463 """
464 Two OSSNodeMigratingNotification notifications are considered equal if they have the same
465 id and are of the same type.
466 """
467 if not isinstance(other, OSSNodeMigratingNotification):
468 return False
469 return self.id == other.id and type(self) is type(other)
471 def __hash__(self) -> int:
472 """
473 Return a hash value for the notification to allow
474 instances to be used in sets and as dictionary keys.
476 Returns:
477 int: Hash value based on notification type and id
478 """
479 return hash((self.__class__.__name__, int(self.id)))
482class OSSNodeMigratedNotification(MaintenanceNotification):
483 """
484 Notification for when a Redis OSS API client is used and a node has completed migrating slots.
486 This notification is received when a node has finished migrating all its slots
487 to other nodes during cluster rebalancing or maintenance operations.
489 Args:
490 id (int): Unique identifier for this notification
491 nodes_to_slots_mapping (Dict[str, List[Dict[str, str]]]): Map of source node address
492 to list of destination mappings. Each destination mapping is a dict with
493 the destination node address as key and the slot range as value.
495 Structure example:
496 {
497 "127.0.0.1:6379": [
498 {"127.0.0.1:6380": "1-100"},
499 {"127.0.0.1:6381": "101-200"}
500 ],
501 "127.0.0.1:6382": [
502 {"127.0.0.1:6383": "201-300"}
503 ]
504 }
506 Where:
507 - Key (str): Source node address in "host:port" format
508 - Value (List[Dict[str, str]]): List of destination mappings where each dict
509 contains destination node address as key and slot range as value
510 """
512 DEFAULT_TTL = 120
514 def __init__(
515 self,
516 id: int,
517 nodes_to_slots_mapping: Dict[str, List[Dict[str, str]]],
518 ):
519 super().__init__(id, OSSNodeMigratedNotification.DEFAULT_TTL)
520 self.nodes_to_slots_mapping = nodes_to_slots_mapping
522 def __repr__(self) -> str:
523 expiry_time = self.creation_time + self.ttl
524 remaining = max(0, expiry_time - time.monotonic())
525 return (
526 f"{self.__class__.__name__}("
527 f"id={self.id}, "
528 f"nodes_to_slots_mapping={self.nodes_to_slots_mapping}, "
529 f"ttl={self.ttl}, "
530 f"creation_time={self.creation_time}, "
531 f"expires_at={expiry_time}, "
532 f"remaining={remaining:.1f}s, "
533 f"expired={self.is_expired()}"
534 f")"
535 )
537 def __eq__(self, other) -> bool:
538 """
539 Two OSSNodeMigratedNotification notifications are considered equal if they have the same
540 id and are of the same type.
541 """
542 if not isinstance(other, OSSNodeMigratedNotification):
543 return False
544 return self.id == other.id and type(self) is type(other)
546 def __hash__(self) -> int:
547 """
548 Return a hash value for the notification to allow
549 instances to be used in sets and as dictionary keys.
551 Returns:
552 int: Hash value based on notification type and id
553 """
554 return hash((self.__class__.__name__, int(self.id)))
557def _is_private_fqdn(host: str) -> bool:
558 """
559 Determine if an FQDN is likely to be internal/private.
561 This uses heuristics based on RFC 952 and RFC 1123 standards:
562 - .local domains (RFC 6762 - Multicast DNS)
563 - .internal domains (common internal convention)
564 - Single-label hostnames (no dots)
565 - Common internal TLDs
567 Args:
568 host (str): The FQDN to check
570 Returns:
571 bool: True if the FQDN appears to be internal/private
572 """
573 host_lower = host.lower().rstrip(".")
575 # Single-label hostnames (no dots) are typically internal
576 if "." not in host_lower:
577 return True
579 # Common internal/private domain patterns
580 internal_patterns = [
581 r"\.local$", # mDNS/Bonjour domains
582 r"\.internal$", # Common internal convention
583 r"\.corp$", # Corporate domains
584 r"\.lan$", # Local area network
585 r"\.intranet$", # Intranet domains
586 r"\.private$", # Private domains
587 ]
589 for pattern in internal_patterns:
590 if re.search(pattern, host_lower):
591 return True
593 # If none of the internal patterns match, assume it's external
594 return False
597notification_types_mapping: dict[type[MaintenanceNotification], str] = {
598 NodeMovingNotification: "MOVING",
599 NodeMigratingNotification: "MIGRATING",
600 NodeMigratedNotification: "MIGRATED",
601 NodeFailingOverNotification: "FAILING_OVER",
602 NodeFailedOverNotification: "FAILED_OVER",
603 OSSNodeMigratingNotification: "SMIGRATING",
604 OSSNodeMigratedNotification: "SMIGRATED",
605}
608def add_debug_log_for_notification(
609 connection: "MaintNotificationsAbstractConnection",
610 notification: Union[str, MaintenanceNotification],
611):
612 if logger.isEnabledFor(logging.DEBUG):
613 logger.debug(
614 f"Handling maintenance notification: {notification}, "
615 f"with connection: {connection}, "
616 f"{connection.extract_connection_details() if connection else 'no connection'}",
617 )
620class MaintNotificationsConfig:
621 """
622 Configuration class for maintenance notifications handling behaviour. Notifications are received through
623 push notifications.
625 This class defines how the Redis client should react to different push notifications
626 such as node moving, migrations, etc. in a Redis cluster.
628 """
630 def __init__(
631 self,
632 enabled: Union[bool, Literal["auto"]] = "auto",
633 proactive_reconnect: bool = True,
634 relaxed_timeout: Optional[Number] = 10,
635 endpoint_type: Optional[EndpointType] = None,
636 event_dispatcher: Optional[EventDispatcherInterface] = None,
637 ):
638 """
639 Initialize a new MaintNotificationsConfig.
641 Args:
642 enabled (bool | "auto"): Controls maintenance notifications handling behavior.
643 - True: The CLIENT MAINT_NOTIFICATIONS command must succeed during connection setup,
644 otherwise a ResponseError is raised.
645 - "auto": The CLIENT MAINT_NOTIFICATIONS command is attempted but failures are
646 gracefully handled - a warning is logged and normal operation continues.
647 - False: Maintenance notifications are completely disabled.
648 Defaults to "auto".
649 proactive_reconnect (bool): Whether to proactively reconnect when a node is replaced.
650 Defaults to True.
651 relaxed_timeout (Number): The relaxed timeout to use for the connection during maintenance.
652 If -1 is provided - the relaxed timeout is disabled. If None is provided - the
653 affected operations become blocking. Defaults to 10.
654 endpoint_type (Optional[EndpointType]): Override for the endpoint type to use in CLIENT MAINT_NOTIFICATIONS.
655 If None, the endpoint type will be automatically determined based on the host and TLS configuration.
656 Defaults to None.
657 event_dispatcher (Optional[EventDispatcherInterface]): Dispatcher that receives a
658 ``MaintenanceStartedEvent`` when a notification starts relaxing timeouts and a
659 ``MaintenanceCompletedEvent`` when the relaxation is reverted. Dispatched
660 synchronously from the handlers, so listeners must not block. If None,
661 no events are dispatched. Defaults to None.
663 Raises:
664 ValueError: If endpoint_type is provided but is not a valid endpoint type.
665 """
666 self.enabled = enabled
667 self.relaxed_timeout = relaxed_timeout
668 self.proactive_reconnect = proactive_reconnect
669 self.endpoint_type = endpoint_type
670 self.event_dispatcher = event_dispatcher
672 def __repr__(self) -> str:
673 return (
674 f"{self.__class__.__name__}("
675 f"enabled={self.enabled}, "
676 f"proactive_reconnect={self.proactive_reconnect}, "
677 f"relaxed_timeout={self.relaxed_timeout}, "
678 f"endpoint_type={self.endpoint_type!r}, "
679 f"event_dispatcher={self.event_dispatcher!r}"
680 f")"
681 )
683 def is_relaxed_timeouts_enabled(self) -> bool:
684 """
685 Check if the relaxed_timeout is enabled. The '-1' value is used to disable the relaxed_timeout.
686 If relaxed_timeout is set to None, it will make the operation blocking
687 and waiting until any response is received.
689 Returns:
690 True if the relaxed_timeout is enabled, False otherwise.
691 """
692 return self.relaxed_timeout != -1
694 def get_endpoint_type(
695 self,
696 host: str,
697 connection: "MaintNotificationsAbstractConnection | AsyncMaintNotificationsAbstractConnection",
698 ) -> EndpointType:
699 """
700 Determine the appropriate endpoint type for CLIENT MAINT_NOTIFICATIONS command.
702 Logic:
703 1. If endpoint_type is explicitly set, use it
704 2. Otherwise, check the original host from connection.host:
705 - If host is an IP address, use it directly to determine internal-ip vs external-ip
706 - If host is an FQDN, get the resolved IP to determine internal-fqdn vs external-fqdn
708 Args:
709 host: User provided hostname to analyze
710 connection: The connection object to analyze for endpoint type determination
712 Returns:
713 """
715 # If endpoint_type is explicitly set, use it
716 if self.endpoint_type is not None:
717 return self.endpoint_type
719 # Check if the host is an IP address
720 try:
721 ip_addr = ipaddress.ip_address(host)
722 # Host is an IP address - use it directly
723 is_private = ip_addr.is_private
724 return EndpointType.INTERNAL_IP if is_private else EndpointType.EXTERNAL_IP
725 except ValueError:
726 # Host is an FQDN - need to check resolved IP to determine internal vs external
727 pass
729 # Host is an FQDN, get the resolved IP to determine if it's internal or external
730 resolved_ip = connection.get_resolved_ip()
732 if resolved_ip:
733 try:
734 ip_addr = ipaddress.ip_address(resolved_ip)
735 is_private = ip_addr.is_private
736 # Use FQDN types since the original host was an FQDN
737 return (
738 EndpointType.INTERNAL_FQDN
739 if is_private
740 else EndpointType.EXTERNAL_FQDN
741 )
742 except ValueError:
743 # This shouldn't happen since we got the IP from the socket, but fallback
744 pass
746 # Final fallback: use heuristics on the FQDN itself
747 is_private = _is_private_fqdn(host)
748 return EndpointType.INTERNAL_FQDN if is_private else EndpointType.EXTERNAL_FQDN
751_MAINTENANCE_START_NOTIFICATION_TYPES = (
752 NodeMigratingNotification,
753 NodeFailingOverNotification,
754 OSSNodeMigratingNotification,
755)
756_MAINTENANCE_COMPLETED_NOTIFICATION_TYPES = (
757 NodeMigratedNotification,
758 NodeFailedOverNotification,
759 OSSNodeMigratedNotification,
760)
763def _get_maintenance_notification_type(
764 notification: MaintenanceNotification,
765) -> Optional[int]:
766 if notification.__class__ in _MAINTENANCE_START_NOTIFICATION_TYPES:
767 return 1
768 if notification.__class__ in _MAINTENANCE_COMPLETED_NOTIFICATION_TYPES:
769 return 0
770 return None
773def _get_maintenance_notification_name(
774 notification: MaintenanceNotification,
775) -> str:
776 return notification_types_mapping.get(notification.__class__, "")
779def _should_skip_connection_timeout_update(
780 maintenance_state: MaintenanceState,
781 config: MaintNotificationsConfig,
782) -> bool:
783 return (
784 maintenance_state == MaintenanceState.MOVING
785 or not config.is_relaxed_timeouts_enabled()
786 )
789# How long a pooled connection's pending push notifications are read for, in
790# total, when it is handed out. The notifications are already buffered, so a
791# complete frame costs no wait at all; the bound is for a frame only partly
792# received, which would otherwise block the checkout for the connection's socket
793# timeout - or, with a blocking socket, indefinitely.
794PENDING_PUSH_NOTIFICATIONS_READ_TIMEOUT = 1.0
797def _dispatch_maintenance_event(
798 config: MaintNotificationsConfig,
799 event: MaintenanceEvent,
800) -> None:
801 """
802 Dispatch a maintenance event to the dispatcher configured on ``config``,
803 if any. A failing listener is logged and never interrupts the notification
804 handling that emitted the event: the push parser swallows exceptions
805 silently, so letting one out would abort the remaining handling instead.
806 """
807 dispatcher = config.event_dispatcher
808 if dispatcher is None:
809 return
810 try:
811 dispatcher.dispatch(event)
812 except Exception:
813 logger.exception(f"Error dispatching maintenance event: {event}")
816def _build_moving_connection_kwargs(
817 notification: NodeMovingNotification,
818 config: MaintNotificationsConfig,
819) -> dict[str, Any]:
820 kwargs: dict[str, Any] = {
821 "maintenance_state": MaintenanceState.MOVING,
822 "maintenance_notification_hash": hash(notification),
823 }
824 if notification.new_node_host is not None:
825 # the host is not updated if the new node host is None
826 # this happens when the MOVING push notification does not contain
827 # the new node host - in this case we only update the timeouts
828 kwargs["host"] = notification.new_node_host
829 if config.is_relaxed_timeouts_enabled():
830 kwargs.update(
831 {
832 "socket_timeout": config.relaxed_timeout,
833 "socket_connect_timeout": config.relaxed_timeout,
834 }
835 )
836 return kwargs
839def _build_moving_cleanup_connection_kwargs(
840 connection_kwargs: Mapping[str, Any],
841 notification_hash: int,
842) -> Optional[dict[str, Any]]:
843 # if the current maintenance_notification_hash in kwargs is not matching the notification
844 # it means there has been a new moving notification after this one
845 # and we don't need to revert the kwargs yet
846 if connection_kwargs.get("maintenance_notification_hash") != notification_hash:
847 return None
849 return {
850 "maintenance_state": MaintenanceState.NONE,
851 "maintenance_notification_hash": None,
852 "host": connection_kwargs.get("orig_host_address"),
853 "socket_timeout": connection_kwargs.get("orig_socket_timeout"),
854 "socket_connect_timeout": connection_kwargs.get("orig_socket_connect_timeout"),
855 }
858class MaintNotificationsPoolHandler:
859 def __init__(
860 self,
861 pool: "MaintNotificationsAbstractConnectionPool",
862 config: MaintNotificationsConfig,
863 ) -> None:
864 self.pool = pool
865 self.config = config
866 self._processed_notifications = set()
867 self._lock = threading.RLock()
868 self.connection = None
870 def set_connection(self, connection: "MaintNotificationsAbstractConnection"):
871 self.connection = connection
873 def get_handler_for_connection(self):
874 # Copy all data that should be shared between connections
875 # but each connection should have its own pool handler
876 # since each connection can be in a different state
877 copy = MaintNotificationsPoolHandler(self.pool, self.config)
878 copy._processed_notifications = self._processed_notifications
879 copy._lock = self._lock
880 copy.connection = None
881 return copy
883 def remove_expired_notifications(self):
884 with self._lock:
885 for notification in tuple(self._processed_notifications):
886 if notification.is_expired():
887 self._processed_notifications.remove(notification)
889 def handle_notification(self, notification: MaintenanceNotification):
890 self.remove_expired_notifications()
892 if isinstance(notification, NodeMovingNotification):
893 return self.handle_node_moving_notification(notification)
894 else:
895 logger.error(f"Unhandled notification type: {notification}")
897 def handle_node_moving_notification(self, notification: NodeMovingNotification):
898 if (
899 not self.config.proactive_reconnect
900 and not self.config.is_relaxed_timeouts_enabled()
901 ):
902 return
904 with self._lock:
905 if notification in self._processed_notifications:
906 # nothing to do in the connection pool handling
907 # the notification has already been handled or is expired
908 # just return
909 return
911 if logger.isEnabledFor(logging.DEBUG):
912 logger.debug(
913 f"Handling node MOVING notification: {notification}, "
914 f"with connection: {self.connection}, connected to ip "
915 f"{self.connection.get_resolved_ip() if self.connection else None}"
916 )
917 # Get the current connected address - if any
918 # This is the address that is being moved
919 # and we need to handle only connections
920 # connected to the same address
921 moving_address_src = (
922 self.connection.getpeername() if self.connection else None
923 )
925 # Reported before the pool is relaxed: taking over a connection
926 # under maintenance completes that connection's pair, and a
927 # listener must see the handoff's pair open by then rather than
928 # no relaxation at all.
929 self._dispatch_handoff_event(MaintenanceStartedEvent, notification)
931 try:
932 with self.pool._lock:
933 if getattr(self.pool, "set_in_maintenance", False):
934 # Set pool in maintenance mode - executed only if
935 # BlockingConnectionPool is used
936 self.pool.set_in_maintenance(True)
938 # Update maintenance state, timeout and optionally host
939 # address connection settings for matching connections
940 self.pool.update_connections_settings(
941 state=MaintenanceState.MOVING,
942 maintenance_notification_hash=hash(notification),
943 relaxed_timeout=self.config.relaxed_timeout,
944 host_address=notification.new_node_host,
945 matching_address=moving_address_src,
946 matching_pattern="connected_address",
947 update_notification_hash=True,
948 include_free_connections=True,
949 )
951 if self.config.proactive_reconnect:
952 if notification.new_node_host is not None:
953 self.run_proactive_reconnect(moving_address_src)
954 else:
955 threading.Timer(
956 notification.ttl / 2,
957 self.run_proactive_reconnect,
958 args=(moving_address_src,),
959 ).start()
961 # Update config for new connections:
962 # Set state to MOVING
963 # update host
964 # if relax timeouts are enabled - update timeouts
965 self.pool.update_connection_kwargs(
966 **_build_moving_connection_kwargs(notification, self.config)
967 )
969 if getattr(self.pool, "set_in_maintenance", False):
970 self.pool.set_in_maintenance(False)
972 threading.Timer(
973 notification.ttl,
974 self.handle_node_moved_notification,
975 args=(notification,),
976 ).start()
977 except BaseException:
978 # The relaxation reported above was not applied, or has no
979 # cleanup to end it: close its pair
980 self._dispatch_handoff_event(MaintenanceCompletedEvent, notification)
981 raise
983 record_connection_handoff(
984 pool_name=get_pool_name(self.pool),
985 )
987 self._processed_notifications.add(notification)
989 def _dispatch_handoff_event(
990 self,
991 event_type: Type[MaintenanceEvent],
992 notification: NodeMovingNotification,
993 ) -> None:
994 """
995 Report a handoff's relaxation. The events report relaxed timeouts; a
996 handoff handled only for the proactive reconnect relaxes nothing.
997 """
998 if self.config.is_relaxed_timeouts_enabled():
999 _dispatch_maintenance_event(
1000 self.config,
1001 event_type(
1002 connection_pool=self.pool,
1003 connection=self.connection,
1004 state=MaintenanceState.MOVING,
1005 notification=notification,
1006 config=self.config,
1007 ),
1008 )
1010 def run_proactive_reconnect(self, moving_address_src: Optional[str] = None):
1011 """
1012 Run proactive reconnect for the pool.
1013 Active connections are marked for reconnect after they complete the current command.
1014 Inactive connections are disconnected and will be connected on next use.
1015 """
1016 with self._lock:
1017 with self.pool._lock:
1018 # take care for the active connections in the pool
1019 # mark them for reconnect after they complete the current command
1020 self.pool.update_active_connections_for_reconnect(
1021 moving_address_src=moving_address_src,
1022 )
1023 # take care for the inactive connections in the pool
1024 # delete them and create new ones
1025 self.pool.disconnect_free_connections(
1026 moving_address_src=moving_address_src,
1027 )
1029 def handle_node_moved_notification(self, notification: NodeMovingNotification):
1030 """
1031 Handle the cleanup after a node moving notification expires.
1032 """
1033 notification_hash = hash(notification)
1035 with self._lock:
1036 if logger.isEnabledFor(logging.DEBUG):
1037 logger.debug(
1038 f"Reverting temporary changes related to notification: {notification}, "
1039 f"with connection: {self.connection}, connected to ip "
1040 f"{self.connection.get_resolved_ip() if self.connection else None}"
1041 )
1042 kwargs = _build_moving_cleanup_connection_kwargs(
1043 self.pool.connection_kwargs, notification_hash
1044 )
1045 if kwargs is not None:
1046 self.pool.update_connection_kwargs(**kwargs)
1048 with self.pool._lock:
1049 reset_relaxed_timeout = self.config.is_relaxed_timeouts_enabled()
1050 reset_host_address = self.config.proactive_reconnect
1052 self.pool.update_connections_settings(
1053 relaxed_timeout=-1,
1054 state=MaintenanceState.NONE,
1055 maintenance_notification_hash=None,
1056 matching_notification_hash=notification_hash,
1057 matching_pattern="notification_hash",
1058 update_notification_hash=True,
1059 reset_relaxed_timeout=reset_relaxed_timeout,
1060 reset_host_address=reset_host_address,
1061 include_free_connections=True,
1062 )
1064 # A newer MOVING notification has superseded this one when the
1065 # kwargs were not reverted: the pool stays relaxed until it expires.
1066 if kwargs is not None:
1067 self._dispatch_handoff_event(MaintenanceCompletedEvent, notification)
1070class MaintNotificationsConnectionHandler:
1071 # 1 = "starting maintenance" notifications, 0 = "completed maintenance" notifications
1072 _NOTIFICATION_TYPES: dict[type["MaintenanceNotification"], int] = {
1073 NodeMigratingNotification: 1,
1074 NodeFailingOverNotification: 1,
1075 OSSNodeMigratingNotification: 1,
1076 NodeMigratedNotification: 0,
1077 NodeFailedOverNotification: 0,
1078 OSSNodeMigratedNotification: 0,
1079 }
1081 def __init__(
1082 self,
1083 connection: "MaintNotificationsAbstractConnection",
1084 config: MaintNotificationsConfig,
1085 ) -> None:
1086 self.connection = connection
1087 self.config = config
1089 def _get_pool(self) -> Optional["MaintNotificationsAbstractConnectionPool"]:
1090 """
1091 Get the pool from the connection's pool handler, or None for
1092 standalone connections without a pool.
1093 """
1094 pool_handler = getattr(
1095 self.connection, "_maint_notifications_pool_handler", None
1096 )
1097 if pool_handler:
1098 return getattr(pool_handler, "pool", None)
1099 return None
1101 def _get_pool_name(self) -> str:
1102 """
1103 Get the pool name from the connection's pool handler.
1104 Falls back to connection representation if pool is not available.
1105 """
1106 pool = self._get_pool()
1107 if pool:
1108 return get_pool_name(pool)
1109 # Fallback for standalone connections without a pool
1110 return repr(self.connection)
1112 def handle_notification(self, notification: MaintenanceNotification):
1113 # get the notification type by checking its class
1114 # 1 for start, 0 for end notification type, None for unknown
1115 notification_type = _get_maintenance_notification_type(notification)
1116 maint_notification = _get_maintenance_notification_name(notification)
1118 record_maint_notification_count(
1119 server_address=self.connection.host,
1120 server_port=self.connection.port,
1121 network_peer_address=self.connection.host,
1122 network_peer_port=self.connection.port,
1123 maint_notification=maint_notification,
1124 )
1126 if notification_type is None:
1127 logger.error(f"Unhandled notification type: {notification}")
1128 return
1130 if notification_type:
1131 self.handle_maintenance_start_notification(
1132 MaintenanceState.MAINTENANCE, notification
1133 )
1134 else:
1135 self.handle_maintenance_completed_notification(notification=notification)
1137 def handle_maintenance_start_notification(
1138 self, maintenance_state: MaintenanceState, notification: MaintenanceNotification
1139 ):
1140 add_debug_log_for_notification(self.connection, notification)
1142 if _should_skip_connection_timeout_update(
1143 self.connection.maintenance_state, self.config
1144 ):
1145 return
1147 self.connection.maintenance_state = maintenance_state
1148 self.connection.set_tmp_settings(
1149 tmp_relaxed_timeout=self.config.relaxed_timeout
1150 )
1151 # extend the timeout for all created connections
1152 self.connection.update_current_socket_timeout(self.config.relaxed_timeout)
1153 if isinstance(notification, OSSNodeMigratingNotification):
1154 # add the notification id to the set of processed start maint notifications
1155 # this is used to skip the unrelaxing of the timeouts if we have received more than
1156 # one start notification before the final end notification
1157 self.connection.add_maint_start_notification(notification.id)
1159 maint_notification = _get_maintenance_notification_name(notification)
1160 record_connection_relaxed_timeout(
1161 connection_name=self._get_pool_name(),
1162 maint_notification=maint_notification,
1163 relaxed=True,
1164 )
1166 _dispatch_maintenance_event(
1167 self.config,
1168 MaintenanceStartedEvent(
1169 connection_pool=self._get_pool(),
1170 connection=self.connection,
1171 state=maintenance_state,
1172 notification=notification,
1173 config=self.config,
1174 ),
1175 )
1177 def handle_maintenance_completed_notification(self, **kwargs: Any) -> None:
1178 # Only reset timeouts if state is not MOVING and relaxed timeouts are enabled
1179 if _should_skip_connection_timeout_update(
1180 self.connection.maintenance_state, self.config
1181 ):
1182 return
1184 # A completion can reach a connection whose start went to another one,
1185 # or none at all; only a connection that was relaxed has a start event
1186 # to pair the completion with
1187 was_relaxed = self.connection.maintenance_state == MaintenanceState.MAINTENANCE
1189 notification = None
1190 if kwargs.get("notification"):
1191 notification = kwargs["notification"]
1192 add_debug_log_for_notification(
1193 self.connection, notification if notification else "MAINTENANCE_COMPLETED"
1194 )
1195 self.connection.reset_tmp_settings(reset_relaxed_timeout=True)
1196 # Maintenance completed - reset the connection
1197 # timeouts by providing -1 as the relaxed timeout
1198 self.connection.update_current_socket_timeout(-1)
1199 self.connection.maintenance_state = MaintenanceState.NONE
1200 # reset the sets that keep track of received start maint
1201 # notifications and skipped end maint notifications
1202 self.connection.reset_received_notifications()
1204 if notification:
1205 maint_notification = _get_maintenance_notification_name(notification)
1206 record_connection_relaxed_timeout(
1207 connection_name=self._get_pool_name(),
1208 maint_notification=maint_notification,
1209 relaxed=False,
1210 )
1212 if was_relaxed:
1213 _dispatch_maintenance_event(
1214 self.config,
1215 MaintenanceCompletedEvent(
1216 connection_pool=self._get_pool(),
1217 connection=self.connection,
1218 state=MaintenanceState.MAINTENANCE,
1219 notification=notification,
1220 config=self.config,
1221 ),
1222 )
1225class OSSMaintNotificationsHandler:
1226 def __init__(
1227 self,
1228 cluster_client: "MaintNotificationsAbstractRedisCluster",
1229 config: MaintNotificationsConfig,
1230 ) -> None:
1231 self.cluster_client = cluster_client
1232 self.config = config
1233 self._processed_notifications = set()
1234 self._in_progress = set()
1235 self._lock = threading.RLock()
1237 def remove_expired_notifications(self):
1238 with self._lock:
1239 for notification in tuple(self._processed_notifications):
1240 if notification.is_expired():
1241 self._processed_notifications.remove(notification)
1243 def handle_notification(self, notification: MaintenanceNotification):
1244 if isinstance(notification, OSSNodeMigratedNotification):
1245 self.handle_oss_maintenance_completed_notification(notification)
1246 else:
1247 logger.error(f"Unhandled notification type: {notification}")
1249 def handle_oss_maintenance_completed_notification(
1250 self, notification: OSSNodeMigratedNotification
1251 ):
1252 self.remove_expired_notifications()
1254 with self._lock:
1255 if (
1256 notification in self._in_progress
1257 or notification in self._processed_notifications
1258 ):
1259 # we are already handling this notification or it has already been processed
1260 # we should skip in_progress notification since when we reinitialize the cluster
1261 # we execute a CLUSTER SLOTS command that can use a different connection
1262 # that has also has the notification and we don't want to
1263 # process the same notification twice
1264 #
1265 # This cheap pre-check runs under _lock alone, before the
1266 # topology refresh lock below. The server delivers the same
1267 # SMIGRATED on every connection, so the vast majority of
1268 # arrivals land here; making them wait for an in-flight CLUSTER
1269 # SLOTS round trip just to discard a notification would stall
1270 # arbitrary command threads inside read_response.
1271 return
1273 # Lock order: _initialization_lock BEFORE _lock. initialize() below runs a
1274 # CLUSTER SLOTS round trip while holding _initialization_lock, and the
1275 # response can carry another SMIGRATED push that is handled inline on that
1276 # thread and needs _lock - so a thread holding _lock must never wait for
1277 # _initialization_lock. See the _initialization_lock comment in
1278 # NodesManager.__init__ for the full ordering rule.
1279 with self.cluster_client.nodes_manager._initialization_lock, self._lock:
1280 if (
1281 notification in self._in_progress
1282 or notification in self._processed_notifications
1283 ):
1284 # Re-check now that both locks are held: another thread may have
1285 # handled the notification while we waited for the refresh lock.
1286 return
1288 if logger.isEnabledFor(logging.DEBUG):
1289 logger.debug(f"Handling SMIGRATED notification: {notification}")
1290 self._in_progress.add(notification)
1292 try:
1293 # Extract the information about the src and destination nodes that are affected
1294 # by the maintenance. nodes_to_slots_mapping structure:
1295 # {
1296 # "src_host:port": [
1297 # {"dest_host:port": "slot_range"},
1298 # ...
1299 # ],
1300 # ...
1301 # }
1302 additional_startup_nodes_info = []
1303 affected_nodes = set()
1304 for (
1305 src_address,
1306 dest_mappings,
1307 ) in notification.nodes_to_slots_mapping.items():
1308 src_host, src_port = src_address.rsplit(":", 1)
1309 src_node = self.cluster_client.nodes_manager.get_node(
1310 host=src_host, port=int(src_port)
1311 )
1312 if src_node is not None:
1313 affected_nodes.add(src_node)
1315 for dest_mapping in dest_mappings:
1316 for dest_address in dest_mapping.keys():
1317 dest_host, dest_port = dest_address.rsplit(":", 1)
1318 additional_startup_nodes_info.append(
1319 (dest_host, int(dest_port))
1320 )
1322 # Updates the cluster slots cache with the new slots mapping
1323 # This will also update the nodes cache with the new nodes mapping
1324 self.cluster_client.nodes_manager.initialize(
1325 disconnect_startup_nodes_pools=False,
1326 additional_startup_nodes_info=additional_startup_nodes_info,
1327 )
1329 all_nodes = set(affected_nodes)
1330 all_nodes = all_nodes.union(
1331 self.cluster_client.nodes_manager.nodes_cache.values()
1332 )
1334 for current_node in all_nodes:
1335 if current_node.redis_connection is None:
1336 continue
1337 with current_node.redis_connection.connection_pool._lock:
1338 handoff_recorded = False
1339 if current_node in affected_nodes:
1340 # mark for reconnect all in use connections to the node - this will force them to
1341 # disconnect after they complete their current commands
1342 # Some of them might be used by sub sub and we don't know which ones - so we disconnect
1343 # all in flight connections after they are done with current command execution
1344 for conn in current_node.redis_connection.connection_pool._get_in_use_connections():
1345 add_debug_log_for_notification(
1346 conn, "SMIGRATED - mark for reconnect"
1347 )
1348 conn.mark_for_reconnect()
1350 record_connection_handoff(
1351 pool_name=get_pool_name(
1352 current_node.redis_connection.connection_pool
1353 )
1354 )
1355 handoff_recorded = True
1356 else:
1357 if logger.isEnabledFor(logging.DEBUG):
1358 logger.debug(
1359 f"SMIGRATED: Node {current_node.name} not affected by maintenance, "
1360 f"skipping mark for reconnect"
1361 )
1363 if (
1364 current_node
1365 not in self.cluster_client.nodes_manager.nodes_cache.values()
1366 ):
1367 # disconnect all free connections to the node - this node will be dropped
1368 # from the cluster, so we don't need to revert the timeouts
1369 for conn in current_node.redis_connection.connection_pool._get_free_connections():
1370 conn.disconnect()
1372 # Only record handoff if not already recorded for this node
1373 if not handoff_recorded:
1374 record_connection_handoff(
1375 pool_name=get_pool_name(
1376 current_node.redis_connection.connection_pool
1377 )
1378 )
1380 # mark the notification as processed
1381 self._processed_notifications.add(notification)
1382 finally:
1383 # Release the in-progress reservation. On success the notification
1384 # is also in _processed_notifications (so it won't be re-handled);
1385 # on failure it is not, allowing a later redelivery to retry.
1386 self._in_progress.discard(notification)