Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/redis/cluster.py: 18%
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 logging
2import random
3import socket
4import sys
5import threading
6import time
7import weakref
8from abc import ABC, abstractmethod
9from collections import OrderedDict, defaultdict
10from concurrent.futures import Future, ThreadPoolExecutor
11from contextlib import contextmanager, nullcontext
12from copy import copy
13from enum import Enum
14from functools import partial
15from itertools import chain
16from types import MethodType
17from typing import (
18 TYPE_CHECKING,
19 Any,
20 Callable,
21 Dict,
22 Iterable,
23 List,
24 Literal,
25 Optional,
26 Set,
27 Tuple,
28 Type,
29 Union,
30)
32if TYPE_CHECKING:
33 from redis.keyspace_notifications import ClusterKeyspaceNotifications
35from redis import _himport_exec
36from redis._defaults import DEFAULT_RETRY_BASE, DEFAULT_RETRY_CAP, DEFAULT_RETRY_COUNT
37from redis._parsers import CommandsParser, Encoder
38from redis._parsers.helpers import parse_scan
39from redis.backoff import ExponentialWithJitterBackoff, NoBackoff
40from redis.cache import CacheConfig, CacheFactory, CacheFactoryInterface, CacheInterface
41from redis.client import EMPTY_RESPONSE, CaseInsensitiveDict, PubSub, Redis
42from redis.commands import RedisClusterCommands
43from redis.commands.helpers import list_or_args, parse_pubsub_subscriptions
44from redis.commands.metadata import (
45 _DEFAULT_KEYED_METADATA,
46 _DEFAULT_KEYLESS_METADATA,
47 _METADATA_BY_REQUEST_POLICY,
48 CommandMetadata,
49 CommandPolicies,
50 MetadataResolver,
51 RequestPolicy,
52 ResponsePolicy,
53 StaticMetadataResolver,
54)
55from redis.commands.policies import PolicyResolver, StaticPolicyResolver
56from redis.connection import (
57 Connection,
58 ConnectionPool,
59 parse_url,
60)
61from redis.crc import REDIS_CLUSTER_HASH_SLOTS, key_slot
62from redis.event import (
63 AfterPooledConnectionsInstantiationEvent,
64 AfterPubSubConnectionInstantiationEvent,
65 AfterSlotsCacheRefreshEvent,
66 ClientType,
67 EventDispatcher,
68 EventListenerInterface,
69)
70from redis.exceptions import (
71 AskError,
72 AuthenticationError,
73 AuthorizationError,
74 ClusterDownError,
75 ClusterError,
76 ConnectionError,
77 CrossSlotTransactionError,
78 DataError,
79 ExecAbortError,
80 InvalidPipelineStack,
81 MaxConnectionsError,
82 MovedError,
83 RedisClusterException,
84 RedisClusterUnreachableError,
85 RedisError,
86 ResponseError,
87 SlotNotCoveredError,
88 TimeoutError,
89 TryAgainError,
90 WatchError,
91)
92from redis.himport import HImportRegistry, parse_himport_set_args
93from redis.lock import Lock
94from redis.maint_notifications import (
95 MaintNotificationsConfig,
96 OSSMaintNotificationsHandler,
97)
98from redis.observability.recorder import (
99 record_error_count,
100 record_operation_duration,
101)
102from redis.retry import Retry
103from redis.typing import (
104 ChannelT,
105 FieldT,
106 PubSubHandler,
107 Subscription,
108)
109from redis.utils import (
110 check_protocol_version,
111 deprecated_args,
112 deprecated_function,
113 dict_merge,
114 experimental_method,
115 list_keys_to_dict,
116 merge_result,
117 safe_str,
118 str_if_bytes,
119 truncate_text,
120)
122logger = logging.getLogger(__name__)
125def is_debug_log_enabled():
126 return logger.isEnabledFor(logging.DEBUG)
129def get_node_name(host: str, port: Union[str, int]) -> str:
130 return f"{host}:{port}"
133@deprecated_args(
134 allowed_args=["redis_node"],
135 reason="Use get_connection(redis_node) instead",
136 version="5.3.0",
137)
138def get_connection(redis_node: Redis, *args, **options) -> Connection:
139 return redis_node.connection or redis_node.connection_pool.get_connection()
142def parse_scan_result(command, res, **options):
143 cursors = {}
144 ret = []
145 for node_name, response in res.items():
146 cursor, r = parse_scan(response, **options)
147 cursors[node_name] = cursor
148 ret += r
150 return cursors, ret
153def parse_pubsub_numsub(command, res, **options):
154 numsub_d = OrderedDict()
155 for numsub_tups in res.values():
156 for channel, numsubbed in numsub_tups:
157 try:
158 numsub_d[channel] += numsubbed
159 except KeyError:
160 numsub_d[channel] = numsubbed
162 ret_numsub = [(channel, numsub) for channel, numsub in numsub_d.items()]
163 return ret_numsub
166def parse_cluster_slots(
167 resp: Any, **options: Any
168) -> Dict[Tuple[int, int], Dict[str, Any]]:
169 current_host = options.get("current_host", "")
171 def fix_server(*args: Any) -> Tuple[str, Any]:
172 return str_if_bytes(args[0]) or current_host, args[1]
174 slots = {}
175 for slot in resp:
176 start, end, primary = slot[:3]
177 replicas = slot[3:]
178 slots[start, end] = {
179 "primary": fix_server(*primary),
180 "replicas": [fix_server(*replica) for replica in replicas],
181 }
183 return slots
186def parse_cluster_shards(resp, **options):
187 """
188 Parse CLUSTER SHARDS response.
189 """
190 if isinstance(resp[0], dict):
191 return resp
192 shards = []
193 for x in resp:
194 shard = {"slots": [], "nodes": []}
195 for i in range(0, len(x[1]), 2):
196 shard["slots"].append((x[1][i], (x[1][i + 1])))
197 nodes = x[3]
198 for node in nodes:
199 dict_node = {}
200 for i in range(0, len(node), 2):
201 dict_node[node[i]] = node[i + 1]
202 shard["nodes"].append(dict_node)
203 shards.append(shard)
205 return shards
208def parse_cluster_shards_with_str_keys(resp, **options):
209 """
210 Parse CLUSTER SHARDS with string top-level structural keys.
212 RESP2 parsing exposes top-level shard keys as ``"slots"``/``"nodes"``
213 while node attribute keys keep the connection's decoded/raw form. RESP3 can
214 return top-level shard dictionaries directly, so normalize only the
215 structural shard keys and preserve nested node dictionaries as delivered.
216 """
217 if not resp:
218 return resp
219 if not isinstance(resp[0], dict):
220 return parse_cluster_shards(resp, **options)
222 shards = []
223 for shard_resp in resp:
224 slots = shard_resp.get(b"slots", shard_resp.get("slots", []))
225 nodes = shard_resp.get(b"nodes", shard_resp.get("nodes", []))
226 shard = {
227 "slots": [
228 tuple(slot) if isinstance(slot, list) else slot for slot in slots
229 ],
230 "nodes": [dict(node) if isinstance(node, dict) else node for node in nodes],
231 }
232 shards.append(shard)
233 return shards
236def parse_cluster_shards_unified(resp, **options):
237 """
238 Parse CLUSTER SHARDS into the approved unified shape.
240 Top-level shard keys and nested node attribute keys are strings for both
241 RESP2 and RESP3 wire responses.
242 """
243 if not resp:
244 return resp
245 if isinstance(resp[0], dict):
246 shards = []
247 for shard_resp in resp:
248 slots = shard_resp.get(b"slots", shard_resp.get("slots", []))
249 nodes = shard_resp.get(b"nodes", shard_resp.get("nodes", []))
250 shard = {
251 "slots": slots,
252 "nodes": [
253 {str_if_bytes(k): v for k, v in node.items()}
254 if isinstance(node, dict)
255 else node
256 for node in nodes
257 ],
258 }
259 shards.append(shard)
260 return shards
262 shards = []
263 for x in resp:
264 shard = {"slots": [], "nodes": []}
265 for i in range(0, len(x[1]), 2):
266 shard["slots"].append((x[1][i], x[1][i + 1]))
267 nodes = x[3]
268 for node in nodes:
269 dict_node = {}
270 for i in range(0, len(node), 2):
271 dict_node[str_if_bytes(node[i])] = node[i + 1]
272 shard["nodes"].append(dict_node)
273 shards.append(shard)
274 return shards
277def parse_cluster_myshardid(resp, **options):
278 """
279 Parse CLUSTER MYSHARDID response.
280 """
281 return resp.decode("utf-8")
284PRIMARY = "primary"
285REPLICA = "replica"
286SLOT_ID = "slot-id"
288REDIS_ALLOWED_KEYS = (
289 "connection_class",
290 "connection_pool",
291 "connection_pool_class",
292 "client_name",
293 "credential_provider",
294 "db",
295 "decode_responses",
296 "encoding",
297 "encoding_errors",
298 "host",
299 "driver_info",
300 "lib_name",
301 "lib_version",
302 "max_connections",
303 "nodes_flag",
304 "redis_connect_func",
305 "password",
306 "port",
307 "timeout",
308 "queue_class",
309 "retry",
310 "retry_on_timeout",
311 "protocol",
312 "legacy_responses",
313 "socket_connect_timeout",
314 "socket_keepalive",
315 "socket_keepalive_options",
316 "socket_read_size",
317 "socket_timeout",
318 "ssl",
319 "ssl_ca_certs",
320 "ssl_ca_data",
321 "ssl_ca_path",
322 "ssl_certfile",
323 "ssl_cert_reqs",
324 "ssl_include_verify_flags",
325 "ssl_exclude_verify_flags",
326 "ssl_keyfile",
327 "ssl_password",
328 "ssl_check_hostname",
329 "unix_socket_path",
330 "username",
331 "cache",
332 "cache_config",
333 "maint_notifications_config",
334)
335KWARGS_DISABLED_KEYS = ("host", "port", "retry")
338def cleanup_kwargs(**kwargs):
339 """
340 Remove unsupported or disabled keys from kwargs
341 """
342 connection_kwargs = {
343 k: v
344 for k, v in kwargs.items()
345 if k in REDIS_ALLOWED_KEYS and k not in KWARGS_DISABLED_KEYS
346 }
348 return connection_kwargs
351class MaintNotificationsAbstractRedisCluster:
352 """
353 Abstract class for handling maintenance notifications logic.
354 This class is expected to be used as base class together with RedisCluster.
356 This class is intended to be used with multiple inheritance!
358 All logic related to maintenance notifications is encapsulated in this class.
359 """
361 def __init__(
362 self,
363 maint_notifications_config: Optional[MaintNotificationsConfig],
364 **kwargs,
365 ):
366 # Initialize maintenance notifications.
367 # The RESP3 requirement is validated in RedisCluster.__init__ before the
368 # NodesManager is constructed; this mixin is only ever run from there, so
369 # the config it receives has already been validated.
370 is_protocol_supported = check_protocol_version(kwargs.get("protocol"), 3)
372 if maint_notifications_config is None and is_protocol_supported:
373 maint_notifications_config = MaintNotificationsConfig()
375 self.maint_notifications_config = maint_notifications_config
377 if self.maint_notifications_config and self.maint_notifications_config.enabled:
378 self._oss_cluster_maint_notifications_handler = (
379 OSSMaintNotificationsHandler(self, self.maint_notifications_config)
380 )
381 # Update connection kwargs for all future nodes connections
382 self._update_connection_kwargs_for_maint_notifications(
383 self._oss_cluster_maint_notifications_handler
384 )
385 # Update existing nodes connections - they are created as part of the RedisCluster constructor
386 for node in self.get_nodes():
387 if node.redis_connection is None:
388 continue
389 node.redis_connection.connection_pool.update_maint_notifications_config(
390 self.maint_notifications_config,
391 oss_cluster_maint_notifications_handler=self._oss_cluster_maint_notifications_handler,
392 )
393 else:
394 self._oss_cluster_maint_notifications_handler = None
396 def _update_connection_kwargs_for_maint_notifications(
397 self, oss_cluster_maint_notifications_handler: OSSMaintNotificationsHandler
398 ):
399 """
400 Update the connection kwargs for all future connections.
401 """
402 self.nodes_manager.connection_kwargs.update(
403 {
404 "oss_cluster_maint_notifications_handler": oss_cluster_maint_notifications_handler,
405 }
406 )
409class AbstractRedisCluster:
410 RedisClusterRequestTTL = 16
412 PRIMARIES = "primaries"
413 REPLICAS = "replicas"
414 ALL_NODES = "all"
415 RANDOM = "random"
416 DEFAULT_NODE = "default-node"
418 NODE_FLAGS = {PRIMARIES, REPLICAS, ALL_NODES, RANDOM, DEFAULT_NODE}
420 COMMAND_FLAGS = dict_merge(
421 list_keys_to_dict(
422 [
423 "ACL CAT",
424 "ACL DELUSER",
425 "ACL DRYRUN",
426 "ACL GENPASS",
427 "ACL GETUSER",
428 "ACL HELP",
429 "ACL LIST",
430 "ACL LOG",
431 "ACL LOAD",
432 "ACL SAVE",
433 "ACL SETUSER",
434 "ACL USERS",
435 "ACL WHOAMI",
436 "AUTH",
437 "CLIENT LIST",
438 "CLIENT SETINFO",
439 "CLIENT SETNAME",
440 "CLIENT GETNAME",
441 "CONFIG SET",
442 "CONFIG REWRITE",
443 "CONFIG RESETSTAT",
444 "TIME",
445 "PUBSUB CHANNELS",
446 "PUBSUB NUMPAT",
447 "PUBSUB NUMSUB",
448 "PUBSUB SHARDCHANNELS",
449 "PUBSUB SHARDNUMSUB",
450 "PING",
451 "INFO",
452 "SHUTDOWN",
453 "KEYS",
454 "DBSIZE",
455 "BGSAVE",
456 "SLOWLOG GET",
457 "SLOWLOG LEN",
458 "SLOWLOG RESET",
459 "WAIT",
460 "WAITAOF",
461 "SAVE",
462 "MEMORY PURGE",
463 "MEMORY MALLOC-STATS",
464 "MEMORY STATS",
465 "LASTSAVE",
466 "CLIENT TRACKINGINFO",
467 "CLIENT PAUSE",
468 "CLIENT UNPAUSE",
469 "CLIENT UNBLOCK",
470 "CLIENT ID",
471 "CLIENT REPLY",
472 "CLIENT GETREDIR",
473 "CLIENT INFO",
474 "CLIENT KILL",
475 "READONLY",
476 "CLUSTER INFO",
477 "CLUSTER MEET",
478 "CLUSTER MYSHARDID",
479 "CLUSTER NODES",
480 "CLUSTER REPLICAS",
481 "CLUSTER RESET",
482 "CLUSTER SET-CONFIG-EPOCH",
483 "CLUSTER SLOTS",
484 "CLUSTER SHARDS",
485 "CLUSTER COUNT-FAILURE-REPORTS",
486 "CLUSTER KEYSLOT",
487 "COMMAND",
488 "COMMAND COUNT",
489 "COMMAND LIST",
490 "COMMAND GETKEYS",
491 "CONFIG GET",
492 "DEBUG",
493 "RANDOMKEY",
494 "READONLY",
495 "READWRITE",
496 "TIME",
497 "TFUNCTION LOAD",
498 "TFUNCTION DELETE",
499 "TFUNCTION LIST",
500 "TFCALL",
501 "TFCALLASYNC",
502 "LATENCY HISTORY",
503 "LATENCY LATEST",
504 "LATENCY RESET",
505 "MODULE LIST",
506 "MODULE LOAD",
507 "MODULE UNLOAD",
508 "MODULE LOADEX",
509 ],
510 DEFAULT_NODE,
511 ),
512 list_keys_to_dict(
513 [
514 "FLUSHALL",
515 "FLUSHDB",
516 "FUNCTION DELETE",
517 "FUNCTION FLUSH",
518 "FUNCTION LIST",
519 "FUNCTION LOAD",
520 "FUNCTION RESTORE",
521 "SCAN",
522 "SCRIPT EXISTS",
523 "SCRIPT FLUSH",
524 "SCRIPT LOAD",
525 ],
526 PRIMARIES,
527 ),
528 list_keys_to_dict(["FUNCTION DUMP"], RANDOM),
529 list_keys_to_dict(
530 [
531 "CLUSTER COUNTKEYSINSLOT",
532 "CLUSTER DELSLOTS",
533 "CLUSTER DELSLOTSRANGE",
534 "CLUSTER GETKEYSINSLOT",
535 "CLUSTER SETSLOT",
536 ],
537 SLOT_ID,
538 ),
539 )
541 SEARCH_COMMANDS = (
542 [
543 "FT.CREATE",
544 "FT.SEARCH",
545 "FT.AGGREGATE",
546 "FT.EXPLAIN",
547 "FT.EXPLAINCLI",
548 "FT,PROFILE",
549 "FT.ALTER",
550 "FT.DROPINDEX",
551 "FT.ALIASADD",
552 "FT.ALIASUPDATE",
553 "FT.ALIASDEL",
554 "FT.ALIASLIST",
555 "FT.TAGVALS",
556 "FT.SUGADD",
557 "FT.SUGGET",
558 "FT.SUGDEL",
559 "FT.SUGLEN",
560 "FT.SYNUPDATE",
561 "FT.SYNDUMP",
562 "FT.SPELLCHECK",
563 "FT.DICTADD",
564 "FT.DICTDEL",
565 "FT.DICTDUMP",
566 "FT.INFO",
567 "FT._LIST",
568 "FT.CONFIG",
569 "FT.ADD",
570 "FT.DEL",
571 "FT.DROP",
572 "FT.GET",
573 "FT.MGET",
574 "FT.SYNADD",
575 ],
576 )
578 CLUSTER_COMMANDS_RESPONSE_CALLBACKS = {
579 "CLUSTER SLOTS": parse_cluster_slots,
580 "CLUSTER SHARDS": parse_cluster_shards,
581 "CLUSTER MYSHARDID": parse_cluster_myshardid,
582 }
584 RESULT_CALLBACKS = dict_merge(
585 list_keys_to_dict(["PUBSUB NUMSUB", "PUBSUB SHARDNUMSUB"], parse_pubsub_numsub),
586 list_keys_to_dict(
587 ["PUBSUB NUMPAT"], lambda command, res: sum(list(res.values()))
588 ),
589 list_keys_to_dict(
590 ["KEYS", "PUBSUB CHANNELS", "PUBSUB SHARDCHANNELS"], merge_result
591 ),
592 list_keys_to_dict(
593 [
594 "PING",
595 "CONFIG SET",
596 "CONFIG REWRITE",
597 "CONFIG RESETSTAT",
598 "CLIENT SETNAME",
599 "BGSAVE",
600 "SLOWLOG RESET",
601 "SAVE",
602 "MEMORY PURGE",
603 "CLIENT PAUSE",
604 "CLIENT UNPAUSE",
605 ],
606 lambda command, res: all(res.values()) if isinstance(res, dict) else res,
607 ),
608 list_keys_to_dict(
609 ["DBSIZE", "WAIT"],
610 lambda command, res: sum(res.values()) if isinstance(res, dict) else res,
611 ),
612 list_keys_to_dict(
613 ["CLIENT UNBLOCK"], lambda command, res: 1 if sum(res.values()) > 0 else 0
614 ),
615 list_keys_to_dict(["SCAN"], parse_scan_result),
616 list_keys_to_dict(
617 ["SCRIPT LOAD"], lambda command, res: list(res.values()).pop()
618 ),
619 list_keys_to_dict(
620 ["SCRIPT EXISTS"], lambda command, res: [all(k) for k in zip(*res.values())]
621 ),
622 list_keys_to_dict(["SCRIPT FLUSH"], lambda command, res: all(res.values())),
623 )
625 ERRORS_ALLOW_RETRY = (
626 ConnectionError,
627 TimeoutError,
628 ClusterDownError,
629 SlotNotCoveredError,
630 )
632 def replace_default_node(self, target_node: "ClusterNode" = None) -> None:
633 """Replace the default cluster node.
634 A random cluster node will be chosen if target_node isn't passed, and primaries
635 will be prioritized. The default node will not be changed if there are no other
636 nodes in the cluster.
638 Args:
639 target_node (ClusterNode, optional): Target node to replace the default
640 node. Defaults to None.
641 """
642 if target_node:
643 self.nodes_manager.default_node = target_node
644 else:
645 curr_node = self.get_default_node()
646 primaries = [node for node in self.get_primaries() if node != curr_node]
647 if primaries:
648 # Choose a primary if the cluster contains different primaries
649 self.nodes_manager.default_node = random.choice(primaries)
650 else:
651 # Otherwise, choose a primary if the cluster contains different primaries
652 replicas = [node for node in self.get_replicas() if node != curr_node]
653 if replicas:
654 self.nodes_manager.default_node = random.choice(replicas)
657class RedisCluster(
658 AbstractRedisCluster, MaintNotificationsAbstractRedisCluster, RedisClusterCommands
659):
660 # Type discrimination marker for @overload self-type pattern
661 _is_async_client: Literal[False] = False
663 @classmethod
664 def from_url(cls, url: str, **kwargs: Any) -> "RedisCluster":
665 """
666 Return a Redis client object configured from the given URL
668 For example::
670 redis://[[username]:[password]]@localhost:6379/0
671 rediss://[[username]:[password]]@localhost:6379/0
672 unix://[username@]/path/to/socket.sock?db=0[&password=password]
674 Three URL schemes are supported:
676 - `redis://` creates a TCP socket connection. See more at:
677 <https://www.iana.org/assignments/uri-schemes/prov/redis>
678 - `rediss://` creates a SSL wrapped TCP socket connection. See more at:
679 <https://www.iana.org/assignments/uri-schemes/prov/rediss>
680 - ``unix://``: creates a Unix Domain Socket connection.
682 The username, password, hostname and path are passed through
683 urllib.parse.unquote in order to replace any percent-encoded values
684 with their corresponding characters. Querystring values are decoded
685 by urllib.parse.parse_qs and are not unquoted again.
687 There are several ways to specify a database number. The first value
688 found will be used:
690 1. A ``db`` querystring option, e.g. redis://localhost?db=0
691 2. If using the redis:// or rediss:// schemes, the path argument
692 of the url, e.g. redis://localhost/0
693 3. A ``db`` keyword argument to this function.
695 If none of these options are specified, the default db=0 is used.
697 All querystring options are cast to their appropriate Python types.
698 Boolean arguments can be specified with string values "True"/"False"
699 or "Yes"/"No". Values that cannot be properly cast cause a
700 ``ValueError`` to be raised. Once parsed, the querystring arguments
701 and keyword arguments are passed to the ``ConnectionPool``'s
702 class initializer. In the case of conflicting arguments, querystring
703 arguments always win.
705 """
706 return cls(url=url, **kwargs)
708 @deprecated_args(
709 args_to_warn=["read_from_replicas"],
710 reason="Please configure the 'load_balancing_strategy' instead",
711 version="5.3.0",
712 )
713 @deprecated_args(
714 args_to_warn=[
715 "cluster_error_retry_attempts",
716 ],
717 reason="Please configure the 'retry' object instead",
718 version="6.0.0",
719 )
720 def __init__(
721 self,
722 host: Optional[str] = None,
723 port: int = 6379,
724 startup_nodes: Optional[List["ClusterNode"]] = None,
725 cluster_error_retry_attempts: int = DEFAULT_RETRY_COUNT,
726 retry: Optional["Retry"] = None,
727 require_full_coverage: bool = True,
728 reinitialize_steps: int = 5,
729 read_from_replicas: bool = False,
730 load_balancing_strategy: Optional["LoadBalancingStrategy"] = None,
731 dynamic_startup_nodes: bool = True,
732 url: Optional[str] = None,
733 address_remap: Optional[Callable[[Tuple[str, int]], Tuple[str, int]]] = None,
734 cache: Optional[CacheInterface] = None,
735 cache_config: Optional[CacheConfig] = None,
736 event_dispatcher: Optional[EventDispatcher] = None,
737 policy_resolver: Optional[PolicyResolver] = None,
738 maint_notifications_config: Optional[MaintNotificationsConfig] = None,
739 metadata_resolver: Optional[MetadataResolver] = None,
740 **kwargs,
741 ):
742 """
743 Initialize a new RedisCluster client.
745 :param startup_nodes:
746 List of nodes from which initial bootstrapping can be done
747 :param host:
748 Can be used to point to a startup node
749 :param port:
750 Can be used to point to a startup node
751 :param require_full_coverage:
752 When set to False (default value): the client will not require a
753 full coverage of the slots. However, if not all slots are covered,
754 and at least one node has 'cluster-require-full-coverage' set to
755 'yes,' the server will throw a ClusterDownError for some key-based
756 commands. See -
757 https://redis.io/topics/cluster-tutorial#redis-cluster-configuration-parameters
758 When set to True: all slots must be covered to construct the
759 cluster client. If not all slots are covered, RedisClusterException
760 will be thrown.
761 :param read_from_replicas:
762 @deprecated - please use load_balancing_strategy instead
763 Enable read from replicas in READONLY mode. You can read possibly
764 stale data.
765 When set to true, read commands will be assigned between the
766 primary and its replications in a Round-Robin manner.
767 :param load_balancing_strategy:
768 Enable read from replicas in READONLY mode and defines the load balancing
769 strategy that will be used for cluster node selection.
770 The data read from replicas is eventually consistent with the data in primary nodes.
771 :param dynamic_startup_nodes:
772 Set the RedisCluster's startup nodes to all of the discovered nodes.
773 If true (default value), the cluster's discovered nodes will be used to
774 determine the cluster nodes-slots mapping in the next topology refresh.
775 It will remove the initial passed startup nodes if their endpoints aren't
776 listed in the CLUSTER SLOTS output.
777 If you use dynamic DNS endpoints for startup nodes but CLUSTER SLOTS lists
778 specific IP addresses, it is best to set it to false.
779 :param cluster_error_retry_attempts:
780 @deprecated - Please configure the 'retry' object instead
781 In case 'retry' object is set - this argument is ignored!
783 Number of times to retry before raising an error when
784 :class:`~.TimeoutError` or :class:`~.ConnectionError`, :class:`~.SlotNotCoveredError` or
785 :class:`~.ClusterDownError` are encountered
786 :param retry:
787 A retry object that defines the retry strategy and the number of
788 retries for the cluster client.
789 In current implementation for the cluster client (starting form redis-py version 6.0.0)
790 the retry object is not yet fully utilized, instead it is used just to determine
791 the number of retries for the cluster client.
792 In the future releases the retry object will be used to handle the cluster client retries!
793 :param reinitialize_steps:
794 Specifies the number of MOVED errors that need to occur before
795 reinitializing the whole cluster topology. If a MOVED error occurs
796 and the cluster does not need to be reinitialized on this current
797 error handling, only the MOVED slot will be patched with the
798 redirected node.
799 To reinitialize the cluster on every MOVED error, set
800 reinitialize_steps to 1.
801 To avoid reinitializing the cluster on moved errors, set
802 reinitialize_steps to 0.
803 :param address_remap:
804 An optional callable which, when provided with an internal network
805 address of a node, e.g. a `(host, port)` tuple, will return the address
806 where the node is reachable. This can be used to map the addresses at
807 which the nodes _think_ they are, to addresses at which a client may
808 reach them, such as when they sit behind a proxy.
810 :param policy_resolver:
811 Decides the request/response policies each command is routed by - see
812 `redis.commands.policies.PolicyResolver`. Defaults to a
813 `StaticPolicyResolver` built for this client, which resolves the command
814 metadata this library ships. A resolver built from a live `COMMAND` reply is a
815 snapshot of the server it was read from, so give each client its own rather
816 than sharing one across clients on different servers.
817 This is the narrow routing view of `metadata_resolver`, which supersedes it:
818 prefer `metadata_resolver`, which serves routing and every other
819 command-metadata consumer from one object. When both are given this one still
820 decides which nodes a command targets, so that its 7.1.0 behavior does not
821 move. It does not decide anything the routing view cannot express: replica
822 safety and client-side-cache eligibility keep coming from `metadata_resolver`,
823 because a `CommandPolicies` record carries no `is_readonly` flag to answer them
824 with.
825 :param metadata_resolver:
826 Serves the command metadata this client reads - see
827 `redis.commands.metadata.MetadataResolver`. Routing is derived from it, and it
828 is handed to every node's client, where it also decides which commands are
829 eligible for client-side caching. Defaults to a `StaticMetadataResolver` built
830 for this client, which resolves the command metadata this library ships; the
831 library never reads `COMMAND` on its own behalf for this, so the default adds no
832 round trips. Resolvers chain through `with_fallback`, first match wins, so one
833 placed in front of `StaticMetadataResolver` overrides the commands it carries
834 while the static records answer for everything else. To decide eligibility and
835 routing from the connected server, pass a `DynamicMetadataResolver` built from a
836 live `COMMAND` reply - use it with care, because reading that reply relies on a
837 class in the private `redis._parsers` package. Note that a server-derived resolver
838 decides routing here too, and two families of command route worse from the live
839 reply than from the shipped records. The server tips commands such as `EXISTS` and
840 `DEL` with the `multi_shard` request policy this client does not yet implement. And
841 the `movablekeys` reads - `ZINTER`, `ZUNION`, `ZDIFF`, `ZINTERCARD`, `SINTERCARD`,
842 `XREAD` - report their keys only in their key specs, so the live reply yields
843 keyless policies that send them to an arbitrary node rather than the one holding
844 their keys; the shipped records withhold those policies instead, which is what
845 leaves the client to resolve the keys through `COMMAND GETKEYS`. So pair such a
846 resolver with an explicit `policy_resolver=StaticPolicyResolver()` to keep routing
847 on the shipped records.
848 :param maint_notifications_config:
849 Configures the nodes connections to support maintenance notifications - see
850 `redis.maint_notifications.MaintNotificationsConfig` for details.
851 Only supported with RESP3.
852 If not provided and protocol is RESP3, the maintenance notifications
853 will be enabled by default (logic is included in the NodesManager
854 initialization).
855 :**kwargs:
856 Extra arguments that will be sent into Redis instance when created
857 (See Official redis-py doc for supported kwargs - the only limitation
858 is that you can't provide 'retry' object as part of kwargs.
859 [https://github.com/andymccurdy/redis-py/blob/master/redis/client.py])
860 Some kwargs are not supported and will raise a
861 RedisClusterException:
862 - db (Redis do not support database SELECT in cluster mode)
864 """
865 if startup_nodes is None:
866 startup_nodes = []
868 if "db" in kwargs:
869 # Argument 'db' is not possible to use in cluster mode
870 raise RedisClusterException(
871 "Argument 'db' is not possible to use in cluster mode"
872 )
874 if "retry" in kwargs:
875 # Argument 'retry' is not possible to be used in kwargs when in cluster mode
876 # the kwargs are set to the lower level connections to the cluster nodes
877 # and there we provide retry configuration without retries allowed.
878 # The retries should be handled on cluster client level.
879 raise RedisClusterException(
880 "The 'retry' argument cannot be used in kwargs when running in cluster mode."
881 )
883 # Get the startup node/s
884 from_url = False
885 if url is not None:
886 from_url = True
887 url_options = parse_url(url)
888 if "path" in url_options:
889 raise RedisClusterException(
890 "RedisCluster does not currently support Unix Domain "
891 "Socket connections"
892 )
893 if "db" in url_options and url_options["db"] != 0:
894 # Argument 'db' is not possible to use in cluster mode
895 raise RedisClusterException(
896 "A ``db`` querystring option can only be 0 in cluster mode"
897 )
898 kwargs.update(url_options)
899 host = kwargs.get("host")
900 port = kwargs.get("port", port)
901 startup_nodes.append(ClusterNode(host, port))
902 elif host is not None and port is not None:
903 startup_nodes.append(ClusterNode(host, port))
904 elif len(startup_nodes) == 0:
905 # No startup node was provided
906 raise RedisClusterException(
907 "RedisCluster requires at least one node to discover the "
908 "cluster. Please provide one of the following:\n"
909 "1. host and port, for example:\n"
910 " RedisCluster(host='localhost', port=6379)\n"
911 "2. list of startup nodes, for example:\n"
912 " RedisCluster(startup_nodes=[ClusterNode('localhost', 6379),"
913 " ClusterNode('localhost', 6378)])"
914 )
915 # Update the connection arguments
916 # Whenever a new connection is established, RedisCluster's on_connect
917 # method should be run
918 # If the user passed on_connect function we'll save it and run it
919 # inside the RedisCluster.on_connect() function
920 self.user_on_connect_func = kwargs.pop("redis_connect_func", None)
921 kwargs.update({"redis_connect_func": self.on_connect})
922 kwargs = cleanup_kwargs(**kwargs)
923 if retry:
924 self.retry = retry
925 else:
926 self.retry = Retry(
927 backoff=ExponentialWithJitterBackoff(
928 base=DEFAULT_RETRY_BASE, cap=DEFAULT_RETRY_CAP
929 ),
930 retries=cluster_error_retry_attempts,
931 )
933 self.encoder = Encoder(
934 kwargs.get("encoding", "utf-8"),
935 kwargs.get("encoding_errors", "strict"),
936 kwargs.get("decode_responses", False),
937 )
938 protocol = kwargs.get("protocol", None)
939 if (cache_config or cache) and not check_protocol_version(protocol, 3):
940 raise RedisError("Client caching is only supported with RESP version 3")
942 if (
943 maint_notifications_config
944 and maint_notifications_config.enabled
945 and not check_protocol_version(protocol, 3)
946 ):
947 raise RedisError(
948 "Maintenance notifications are only supported with RESP version 3"
949 )
950 if check_protocol_version(protocol, 3) and maint_notifications_config is None:
951 maint_notifications_config = MaintNotificationsConfig()
953 # Build the client-level HIMPORT registry once (always empty at construction)
954 # and share the same object with every node pool, so the fieldset registry is
955 # shared cluster-wide and runtime himport_prepare mutates one object. It is
956 # handed to the NodesManager and injected onto each node's pool in
957 # create_redis_node; it is deliberately NOT forwarded through connection_kwargs,
958 # so nodes reuse the one shared object rather than each rebuilding their own.
959 self._himport_registry = HImportRegistry()
961 self.command_flags = self.__class__.COMMAND_FLAGS.copy()
962 self.node_flags = self.__class__.NODE_FLAGS.copy()
963 self.read_from_replicas = read_from_replicas
964 self.load_balancing_strategy = load_balancing_strategy
965 self.reinitialize_counter = 0
966 self.reinitialize_steps = reinitialize_steps
967 if event_dispatcher is None:
968 self._event_dispatcher = EventDispatcher()
969 else:
970 self._event_dispatcher = event_dispatcher
971 self.startup_nodes = startup_nodes
973 # Built here rather than defaulted in the signature, so that each client owns its
974 # resolver and the memos it accumulates are released with the client. The one object
975 # is shared with every node's client below, so a cluster resolves command metadata -
976 # routing and cache eligibility both - from a single source of truth.
977 if metadata_resolver is None:
978 self._metadata_resolver: MetadataResolver = StaticMetadataResolver()
979 else:
980 self._metadata_resolver = metadata_resolver
982 self.nodes_manager = NodesManager(
983 startup_nodes=startup_nodes,
984 from_url=from_url,
985 require_full_coverage=require_full_coverage,
986 dynamic_startup_nodes=dynamic_startup_nodes,
987 address_remap=address_remap,
988 cache=cache,
989 cache_config=cache_config,
990 metadata_resolver=self._metadata_resolver,
991 event_dispatcher=self._event_dispatcher,
992 maint_notifications_config=maint_notifications_config,
993 himport_registry=self._himport_registry,
994 **kwargs,
995 )
997 cluster_response_callbacks = dict(
998 self.__class__.CLUSTER_COMMANDS_RESPONSE_CALLBACKS
999 )
1000 legacy_responses = kwargs.get("legacy_responses", True)
1001 protocol = kwargs.get("protocol")
1002 if not legacy_responses:
1003 cluster_response_callbacks["CLUSTER SHARDS"] = parse_cluster_shards_unified
1004 elif protocol is None:
1005 cluster_response_callbacks["CLUSTER SHARDS"] = (
1006 parse_cluster_shards_with_str_keys
1007 )
1008 self.cluster_response_callbacks = CaseInsensitiveDict(
1009 cluster_response_callbacks
1010 )
1011 self.result_callbacks = CaseInsensitiveDict(self.__class__.RESULT_CALLBACKS)
1013 # For backward compatibility, mapping from existing policies to new one
1014 self._command_flags_mapping: dict[str, Union[RequestPolicy, ResponsePolicy]] = {
1015 self.__class__.RANDOM: RequestPolicy.DEFAULT_KEYLESS,
1016 self.__class__.PRIMARIES: RequestPolicy.ALL_SHARDS,
1017 self.__class__.ALL_NODES: RequestPolicy.ALL_NODES,
1018 self.__class__.REPLICAS: RequestPolicy.ALL_REPLICAS,
1019 self.__class__.DEFAULT_NODE: RequestPolicy.DEFAULT_NODE,
1020 SLOT_ID: RequestPolicy.DEFAULT_KEYED,
1021 }
1023 self._policies_callback_mapping: dict[
1024 Union[RequestPolicy, ResponsePolicy], Callable
1025 ] = {
1026 RequestPolicy.DEFAULT_KEYLESS: lambda command_name: [
1027 self.get_keyless_target_node(command_name)
1028 ],
1029 RequestPolicy.DEFAULT_KEYED: lambda command,
1030 *args: self.get_nodes_from_slot(command, *args),
1031 RequestPolicy.DEFAULT_NODE: lambda: [self.get_default_node()],
1032 RequestPolicy.ALL_SHARDS: self.get_primaries,
1033 RequestPolicy.ALL_NODES: self.get_nodes,
1034 RequestPolicy.ALL_REPLICAS: self.get_replicas,
1035 RequestPolicy.MULTI_SHARD: lambda *args,
1036 **kwargs: self._split_multi_shard_command(*args, **kwargs),
1037 RequestPolicy.SPECIAL: self.get_special_nodes,
1038 ResponsePolicy.DEFAULT_KEYLESS: lambda res: res,
1039 ResponsePolicy.DEFAULT_KEYED: lambda res: res,
1040 }
1042 # ``policy_resolver`` is the routing view of a metadata resolver, so the two
1043 # arguments overlap. Resolved by precedence rather than by rejecting the
1044 # combination, because a user migrating from one to the other will legitimately pass
1045 # both: an explicit ``policy_resolver`` - the extension point that shipped in 7.1.0
1046 # - keeps deciding which nodes a command targets, and otherwise those policies are
1047 # derived from the metadata resolver.
1048 #
1049 # The precedence covers the routing view only. Replica safety and cache eligibility
1050 # are read from ``_metadata_resolver`` either way, because the projection a policy
1051 # resolver serves drops the flags they are decided from - a ``CommandPolicies``
1052 # record has no ``is_readonly``. So "ignores metadata_resolver" below means for the
1053 # target-node decision, not for the record as a whole.
1054 if policy_resolver is None:
1055 self._policy_resolver: PolicyResolver = StaticPolicyResolver(
1056 metadata_resolver=self._metadata_resolver
1057 )
1058 else:
1059 self._policy_resolver = policy_resolver
1060 if metadata_resolver is not None:
1061 logger.debug(
1062 "Both policy_resolver and metadata_resolver were given; the nodes a "
1063 "command targets resolve through policy_resolver and ignore "
1064 "metadata_resolver. Replica safety and client-side-cache eligibility "
1065 "still resolve through metadata_resolver."
1066 )
1067 self.commands_parser = CommandsParser(self)
1069 # Node where FT.AGGREGATE command is executed.
1070 self._aggregate_nodes = None
1071 self._lock = threading.RLock()
1073 MaintNotificationsAbstractRedisCluster.__init__(
1074 self, maint_notifications_config, **kwargs
1075 )
1077 def __enter__(self):
1078 return self
1080 def __exit__(self, exc_type, exc_value, traceback):
1081 self.close()
1083 def __del__(self):
1084 try:
1085 self.close()
1086 except Exception:
1087 pass
1089 def disconnect_connection_pools(self):
1090 for node in self.get_nodes():
1091 if node.redis_connection:
1092 try:
1093 node.redis_connection.connection_pool.disconnect()
1094 except OSError:
1095 # Client was already disconnected. do nothing
1096 pass
1098 def on_connect(self, connection):
1099 """
1100 Initialize the connection, authenticate and select a database and send
1101 READONLY if it is set during object initialization.
1102 """
1103 connection.on_connect()
1105 if self.read_from_replicas or self.load_balancing_strategy:
1106 # Sending READONLY command to server to configure connection as
1107 # readonly. Since each cluster node may change its server type due
1108 # to a failover, we should establish a READONLY connection
1109 # regardless of the server type. If this is a primary connection,
1110 # READONLY would not affect executing write commands.
1111 connection.send_command("READONLY")
1112 if str_if_bytes(connection.read_response()) != "OK":
1113 raise ConnectionError("READONLY command failed")
1115 if self.user_on_connect_func is not None:
1116 self.user_on_connect_func(connection)
1118 def get_redis_connection(self, node: "ClusterNode") -> Redis:
1119 if not node.redis_connection:
1120 with self._lock:
1121 if not node.redis_connection:
1122 self.nodes_manager.create_redis_connections([node])
1123 return node.redis_connection
1125 def get_node(self, host=None, port=None, node_name=None):
1126 return self.nodes_manager.get_node(host, port, node_name)
1128 def get_primaries(self):
1129 return self.nodes_manager.get_nodes_by_server_type(PRIMARY)
1131 def get_replicas(self):
1132 return self.nodes_manager.get_nodes_by_server_type(REPLICA)
1134 def get_random_node(self):
1135 return random.choice(list(self.nodes_manager.nodes_cache.values()))
1137 def get_keyless_target_node(self, command_name: str) -> "ClusterNode":
1138 """
1139 Returns the node a keyless command is routed to: a random node when replica reads
1140 are enabled and the command is safe to serve from a replica, a random primary
1141 otherwise.
1143 A replicas-only ``load_balancing_strategy`` is honored by picking from the replicas
1144 alone, so a strategy that asks for replicas cannot land on a primary here. The
1145 strategy is not applied any further than that: the rest of it is an index into one
1146 shard's node list and a round-robin counter kept per primary name, and a keyless
1147 command has no shard to index - so the pick is uniform over the eligible nodes.
1149 Falls back to the whole node set when the cluster has no replicas to pick from,
1150 which is every primary. That is also the answer for the two strategies that
1151 include the primary, and for ``read_from_replicas`` on its own, which is what this
1152 method has returned for a replica-safe command since 7.1.0.
1153 """
1154 replica_safe = (
1155 self.read_from_replicas or self.load_balancing_strategy is not None
1156 ) and self._is_replica_safe(command_name)
1157 if replica_safe:
1158 if self.load_balancing_strategy in _REPLICAS_ONLY_STRATEGIES:
1159 replicas = self.get_replicas()
1160 if replicas:
1161 return random.choice(replicas)
1163 return self.get_random_node()
1165 return self.get_random_primary_node()
1167 @deprecated_function(
1168 version="8.2.0",
1169 reason="Use get_keyless_target_node() instead.",
1170 )
1171 def get_random_primary_or_all_nodes(self, command_name: str) -> "ClusterNode":
1172 """
1173 Returns random primary or all nodes depends on READONLY mode.
1175 Deprecated alias of :meth:`get_keyless_target_node`. Kept so the name that has
1176 been public since 7.1.0 keeps working; it answers from the metadata resolver just
1177 as the new name does.
1178 """
1179 return self.get_keyless_target_node(command_name)
1181 def _is_replica_safe(self, command_name: str) -> bool:
1182 return self._metadata_resolver.is_replica_safe(command_name)
1184 def get_nodes(self):
1185 return list(self.nodes_manager.nodes_cache.values())
1187 def get_node_from_key(self, key, replica=False):
1188 """
1189 Get the node that holds the key's slot.
1190 If replica set to True but the slot doesn't have any replicas, None is
1191 returned.
1192 """
1193 slot = self.keyslot(key)
1194 slot_cache = self.nodes_manager.slots_cache.get(slot)
1195 if slot_cache is None or len(slot_cache) == 0:
1196 raise SlotNotCoveredError(f'Slot "{slot}" is not covered by the cluster.')
1197 if replica and len(self.nodes_manager.slots_cache[slot]) < 2:
1198 return None
1199 elif replica:
1200 node_idx = 1
1201 else:
1202 # primary
1203 node_idx = 0
1205 return slot_cache[node_idx]
1207 def get_default_node(self):
1208 """
1209 Get the cluster's default node
1210 """
1211 return self.nodes_manager.default_node
1213 def get_nodes_from_slot(self, command: str, *args):
1214 """
1215 Returns a list of nodes that hold the specified keys' slots.
1216 """
1217 # get the node that holds the key's slot
1218 slot = self.determine_slot(*args)
1219 replica_safe = (
1220 self.read_from_replicas or self.load_balancing_strategy is not None
1221 ) and self._is_replica_safe(command)
1222 node = self.nodes_manager.get_node_from_slot(
1223 slot,
1224 replica_safe,
1225 self.load_balancing_strategy if replica_safe else None,
1226 )
1227 return [node]
1229 def _split_multi_shard_command(self, *args, **kwargs) -> list[dict]:
1230 """
1231 Splits the command with Multi-Shard policy, to the multiple commands
1232 """
1233 keys = self._get_command_keys(*args)
1234 commands = []
1236 for key in keys:
1237 commands.append(
1238 {
1239 "args": (args[0], key),
1240 "kwargs": kwargs,
1241 }
1242 )
1244 return commands
1246 def get_special_nodes(self) -> Optional[list["ClusterNode"]]:
1247 """
1248 Returns a list of nodes for commands with a special policy.
1249 """
1250 if not self._aggregate_nodes:
1251 raise RedisClusterException(
1252 "Cannot execute FT.CURSOR commands without FT.AGGREGATE"
1253 )
1255 return self._aggregate_nodes
1257 def get_random_primary_node(self) -> "ClusterNode":
1258 """
1259 Returns a random primary node
1260 """
1261 return random.choice(self.get_primaries())
1263 def _evaluate_all_succeeded(self, res):
1264 """
1265 Evaluate the result of a command with ResponsePolicy.ALL_SUCCEEDED
1266 """
1267 first_successful_response = None
1269 if isinstance(res, dict):
1270 for key, value in res.items():
1271 if value:
1272 if first_successful_response is None:
1273 first_successful_response = {key: value}
1274 else:
1275 return {key: False}
1276 else:
1277 for response in res:
1278 if response:
1279 if first_successful_response is None:
1280 # Dynamically resolve type
1281 first_successful_response = type(response)(response)
1282 else:
1283 return type(response)(False)
1285 return first_successful_response
1287 def set_default_node(self, node):
1288 """
1289 Set the default node of the cluster.
1290 :param node: 'ClusterNode'
1291 :return True if the default node was set, else False
1292 """
1293 if node is None or self.get_node(node_name=node.name) is None:
1294 return False
1295 self.nodes_manager.default_node = node
1296 return True
1298 def set_retry(self, retry: Retry) -> None:
1299 self.retry = retry
1301 def monitor(self, target_node=None):
1302 """
1303 Returns a Monitor object for the specified target node.
1304 The default cluster node will be selected if no target node was
1305 specified.
1306 Monitor is useful for handling the MONITOR command to the redis server.
1307 next_command() method returns one command from monitor
1308 listen() method yields commands from monitor.
1309 """
1310 if target_node is None:
1311 target_node = self.get_default_node()
1312 if target_node.redis_connection is None:
1313 raise RedisClusterException(
1314 f"Cluster Node {target_node.name} has no redis_connection"
1315 )
1316 return target_node.redis_connection.monitor()
1318 def pubsub(self, node=None, host=None, port=None, **kwargs):
1319 """
1320 Allows passing a ClusterNode, or host&port, to get a pubsub instance
1321 connected to the specified node
1322 """
1323 return ClusterPubSub(self, node=node, host=host, port=port, **kwargs)
1325 def keyspace_notifications(
1326 self,
1327 key_prefix: Union[str, bytes, None] = None,
1328 ignore_subscribe_messages: bool = True,
1329 ) -> "ClusterKeyspaceNotifications":
1330 """
1331 Return a :class:`~redis.keyspace_notifications.ClusterKeyspaceNotifications`
1332 object for subscribing to keyspace and keyevent notifications across
1333 all primary nodes in the cluster.
1335 Note: Keyspace notifications must be enabled on all Redis cluster nodes
1336 via the ``notify-keyspace-events`` configuration option.
1338 Args:
1339 key_prefix: Optional prefix to filter and strip from keys in
1340 notifications.
1341 ignore_subscribe_messages: If True, subscribe/unsubscribe
1342 confirmations are not returned by
1343 get_message/listen.
1344 """
1345 from redis.keyspace_notifications import ClusterKeyspaceNotifications
1347 return ClusterKeyspaceNotifications(
1348 self,
1349 key_prefix=key_prefix,
1350 ignore_subscribe_messages=ignore_subscribe_messages,
1351 )
1353 def pipeline(self, transaction=None, shard_hint=None):
1354 """
1355 Cluster impl:
1356 Pipelines do not work in cluster mode the same way they
1357 do in normal mode. Create a clone of this object so
1358 that simulating pipelines will work correctly. Each
1359 command will be called directly when used and
1360 when calling execute() will only return the result stack.
1361 """
1362 if shard_hint:
1363 raise RedisClusterException("shard_hint is deprecated in cluster mode")
1365 return ClusterPipeline(
1366 nodes_manager=self.nodes_manager,
1367 commands_parser=self.commands_parser,
1368 startup_nodes=self.nodes_manager.startup_nodes,
1369 result_callbacks=self.result_callbacks,
1370 cluster_response_callbacks=self.cluster_response_callbacks,
1371 read_from_replicas=self.read_from_replicas,
1372 load_balancing_strategy=self.load_balancing_strategy,
1373 reinitialize_steps=self.reinitialize_steps,
1374 retry=self.retry,
1375 lock=self._lock,
1376 transaction=transaction,
1377 # Routing must not change just because the commands go through a pipeline, so the
1378 # pipeline resolves through the same objects the client does.
1379 policy_resolver=self._policy_resolver,
1380 metadata_resolver=self._metadata_resolver,
1381 event_dispatcher=self._event_dispatcher,
1382 )
1384 def lock(
1385 self,
1386 name,
1387 timeout=None,
1388 sleep=0.1,
1389 blocking=True,
1390 blocking_timeout=None,
1391 lock_class=None,
1392 thread_local=True,
1393 raise_on_release_error: bool = True,
1394 ):
1395 """
1396 Return a new Lock object using key ``name`` that mimics
1397 the behavior of threading.Lock.
1399 If specified, ``timeout`` indicates a maximum life for the lock.
1400 By default, it will remain locked until release() is called.
1402 ``sleep`` indicates the amount of time to sleep per loop iteration
1403 when the lock is in blocking mode and another client is currently
1404 holding the lock.
1406 ``blocking`` indicates whether calling ``acquire`` should block until
1407 the lock has been acquired or to fail immediately, causing ``acquire``
1408 to return False and the lock not being acquired. Defaults to True.
1409 Note this value can be overridden by passing a ``blocking``
1410 argument to ``acquire``.
1412 ``blocking_timeout`` indicates the maximum amount of time in seconds to
1413 spend trying to acquire the lock. A value of ``None`` indicates
1414 continue trying forever. ``blocking_timeout`` can be specified as a
1415 float or integer, both representing the number of seconds to wait.
1417 ``lock_class`` forces the specified lock implementation. Note that as
1418 of redis-py 3.0, the only lock class we implement is ``Lock`` (which is
1419 a Lua-based lock). So, it's unlikely you'll need this parameter, unless
1420 you have created your own custom lock class.
1422 ``thread_local`` indicates whether the lock token is placed in
1423 thread-local storage. By default, the token is placed in thread local
1424 storage so that a thread only sees its token, not a token set by
1425 another thread. Consider the following timeline:
1427 time: 0, thread-1 acquires `my-lock`, with a timeout of 5 seconds.
1428 thread-1 sets the token to "abc"
1429 time: 1, thread-2 blocks trying to acquire `my-lock` using the
1430 Lock instance.
1431 time: 5, thread-1 has not yet completed. redis expires the lock
1432 key.
1433 time: 5, thread-2 acquired `my-lock` now that it's available.
1434 thread-2 sets the token to "xyz"
1435 time: 6, thread-1 finishes its work and calls release(). if the
1436 token is *not* stored in thread local storage, then
1437 thread-1 would see the token value as "xyz" and would be
1438 able to successfully release the thread-2's lock.
1440 ``raise_on_release_error`` indicates whether to raise an exception when
1441 the lock is no longer owned when exiting the context manager. By default,
1442 this is True, meaning an exception will be raised. If False, the warning
1443 will be logged and the exception will be suppressed.
1445 In some use cases it's necessary to disable thread local storage. For
1446 example, if you have code where one thread acquires a lock and passes
1447 that lock instance to a worker thread to release later. If thread
1448 local storage isn't disabled in this case, the worker thread won't see
1449 the token set by the thread that acquired the lock. Our assumption
1450 is that these cases aren't common and as such default to using
1451 thread local storage."""
1452 if lock_class is None:
1453 lock_class = Lock
1454 return lock_class(
1455 self,
1456 name,
1457 timeout=timeout,
1458 sleep=sleep,
1459 blocking=blocking,
1460 blocking_timeout=blocking_timeout,
1461 thread_local=thread_local,
1462 raise_on_release_error=raise_on_release_error,
1463 )
1465 def set_response_callback(self, command, callback):
1466 """Set a custom Response Callback"""
1467 self.cluster_response_callbacks[command] = callback
1469 def _resolve_command_policies(
1470 self, *args, target_nodes_specified: bool = False
1471 ) -> Tuple[str, Union[CommandPolicies, CommandMetadata]]:
1472 """
1473 Resolves the policies a command routes and aggregates by.
1475 Returns the name the policies were decided by along with the record, because the
1476 name a command is known by is not always ``args[0]``: a container command is
1477 keyed by both of its words, and the flag tables are keyed in upper case.
1479 The name is normalized before any branch, so one command answers with one name
1480 however it got here. The result callbacks are keyed in upper case, so a name that
1481 kept the caller's spelling on only some paths would fire them on only some paths -
1482 ``execute_command("dbsize")`` would be summed and ``execute_command("dbsize",
1483 target_nodes=...)`` would not.
1485 First choice is the policy resolver. When it does not know the command, the
1486 fallbacks are the command's ``COMMAND_FLAGS`` entry and then whether the command
1487 carries a key.
1488 """
1489 command = args[0].upper()
1490 if len(args) >= 2 and f"{args[0]} {args[1]}".upper() in self.command_flags:
1491 command = f"{args[0]} {args[1]}".upper()
1493 if target_nodes_specified:
1494 # The caller named its targets, so nothing is routed from here - and the
1495 # command's own aggregation must not apply either. A response policy resolved
1496 # for the whole cluster (``ONE_SUCCEEDED``, an ``AGG_*``) would short-circuit
1497 # the loop over the nodes the caller picked, or fold their replies into one.
1498 # Answer with the record that aggregates nothing, and skip the resolver: with
1499 # the targets given, neither of its answers is used.
1500 return command, _DEFAULT_KEYLESS_METADATA
1502 policies = self._policy_resolver.resolve(args[0].lower())
1503 if policies:
1504 return command, policies
1506 command_flag = self.command_flags.get(command)
1507 if command_flag:
1508 if command_flag in self._command_flags_mapping:
1509 return command, _METADATA_BY_REQUEST_POLICY[
1510 self._command_flags_mapping[command_flag]
1511 ]
1512 return command, _DEFAULT_KEYLESS_METADATA
1514 # Unflagged and unresolved, so the command routes by its key. Without a default
1515 # node the topology is not known yet and there is no slot to route by.
1516 if not self.get_default_node():
1517 return command, _DEFAULT_KEYLESS_METADATA
1519 slot = self.determine_slot(*args)
1520 if slot is None:
1521 return command, _DEFAULT_KEYLESS_METADATA
1523 return command, _DEFAULT_KEYED_METADATA
1525 def _determine_nodes(
1526 self, *args, request_policy: Optional[RequestPolicy] = None, **kwargs
1527 ) -> List["ClusterNode"]:
1528 """
1529 Determines a nodes the command should be executed on.
1531 The caller resolves the command's own policy - see
1532 ``_resolve_command_policies`` - so the only decision left here is an explicit
1533 nodes flag, which overrides it.
1534 """
1535 command = args[0].upper()
1536 if len(args) >= 2 and f"{args[0]} {args[1]}".upper() in self.command_flags:
1537 command = f"{args[0]} {args[1]}".upper()
1539 nodes_flag = kwargs.pop("nodes_flag", None)
1540 if nodes_flag and self._is_nodes_flag(nodes_flag):
1541 # nodes flag passed by the user
1542 if nodes_flag in self._command_flags_mapping:
1543 request_policy = self._command_flags_mapping[nodes_flag]
1545 if request_policy is None:
1546 raise RedisClusterException(
1547 f"No targets were found to execute {args} command on"
1548 )
1550 policy_callback = self._policies_callback_mapping[request_policy]
1552 if request_policy == RequestPolicy.DEFAULT_KEYED:
1553 nodes = policy_callback(command, *args)
1554 elif request_policy == RequestPolicy.MULTI_SHARD:
1555 nodes = policy_callback(*args, **kwargs)
1556 elif request_policy == RequestPolicy.DEFAULT_KEYLESS:
1557 nodes = policy_callback(args[0])
1558 else:
1559 nodes = policy_callback()
1561 if args[0].lower() == "ft.aggregate":
1562 self._aggregate_nodes = nodes
1564 return nodes
1566 def _should_reinitialized(self):
1567 # To reinitialize the cluster on every MOVED error,
1568 # set reinitialize_steps to 1.
1569 # To avoid reinitializing the cluster on moved errors, set
1570 # reinitialize_steps to 0.
1571 if self.reinitialize_steps == 0:
1572 return False
1573 else:
1574 return self.reinitialize_counter % self.reinitialize_steps == 0
1576 def keyslot(self, key):
1577 """
1578 Calculate keyslot for a given key.
1579 See Keys distribution model in https://redis.io/topics/cluster-spec
1580 """
1581 k = self.encoder.encode(key)
1582 return key_slot(k)
1584 # HIMPORT orchestration. PREPARE/DISCARD/DISCARDALL mutate the one shared
1585 # HImportRegistry exactly once (every node pool references the same object, so the
1586 # change is visible cluster-wide and applied lazily per node). SET routes by key
1587 # slot to the owning primary and reuses that node's standalone himport_set (lazy
1588 # PREPARE bundled with SET). See ``.agents/himport_client_support_spec.md``.
1590 @property
1591 def himport_registry(self) -> HImportRegistry:
1592 """The cluster-wide HIMPORT fieldset registry (empty if none was declared).
1594 Read-only: the registry is mutated only through the HIMPORT command methods.
1595 """
1596 return self._himport_registry
1598 @experimental_method()
1599 def himport_prepare(self, fieldset_name: str, fields: Iterable[FieldT]) -> bool:
1600 """Declare an HIMPORT fieldset cluster-wide (shared registry, applied lazily)."""
1601 self._himport_registry.prepare(fieldset_name, fields)
1602 return True
1604 @experimental_method()
1605 def himport_discard(self, fieldset_name: str) -> int:
1606 """Remove an HIMPORT fieldset cluster-wide (shared registry, applied lazily)."""
1607 return 1 if self._himport_registry.discard(fieldset_name) else 0
1609 @experimental_method()
1610 def himport_discard_all(self) -> int:
1611 """Remove all HIMPORT fieldsets cluster-wide (shared registry, applied lazily)."""
1612 return self._himport_registry.discard_all()
1614 def _get_command_keys(self, *args):
1615 """
1616 Get the keys in the command. If the command has no keys in it, None is
1617 returned.
1619 NOTE: Due to a bug in redis<7.0, this function does not work properly
1620 for EVAL or EVALSHA when the `numkeys` arg is 0.
1621 - issue: https://github.com/redis/redis/issues/9493
1622 - fix: https://github.com/redis/redis/pull/9733
1624 So, don't use this function with EVAL or EVALSHA.
1626 Raises:
1627 RedisClusterException: If the cluster has no default node to resolve
1628 the keys against, which is the case before the slots cache is
1629 first populated and after the client is closed.
1630 """
1631 default_node = self.get_default_node()
1632 if default_node is None:
1633 # The keys are unknown rather than absent: there is no node to resolve them
1634 # against. Say that, rather than reporting a missing key the caller did
1635 # supply, or raising AttributeError from in here.
1636 #
1637 # The async stack needs no counterpart: its parser holds the node it
1638 # was initialized with and never reads the default node.
1639 raise RedisClusterException(
1640 "The cluster has no default node to resolve the keys of this "
1641 "command against. The client may be closed, or not initialized "
1642 f"yet.\nCommand: {args}"
1643 )
1645 return self.commands_parser.get_keys(default_node.redis_connection, *args)
1647 def determine_slot(self, *args) -> Optional[int]:
1648 """
1649 Figure out what slot to use based on args.
1651 Raises a RedisClusterException if there's a missing key and we can't
1652 determine what slots to map the command to; or, if the keys don't
1653 all map to the same key slot; or, when the keys have to be resolved
1654 through ``_get_command_keys``, if the cluster has no default node to
1655 resolve them against.
1657 Returns the slot as an ``int``, which is what the declared return type has always
1658 promised and what the slot map is keyed by. A command carrying its slot as an
1659 argument - the ``SLOT_ID`` group - is therefore cast rather than returned verbatim,
1660 so a caller that spelled the slot as a string gets a usable slot instead of a
1661 ``KeyError`` from ``get_node_from_slot``.
1662 """
1663 command = args[0]
1664 if self.command_flags.get(command.upper()) == SLOT_ID:
1665 # The command contains the slot ID. The flag table is keyed in upper case, so
1666 # the lookup is normalized - a raw ``execute_command("cluster countkeysinslot",
1667 # ...)`` names the same command as the spelling the command method sends.
1668 return int(args[1])
1670 # Get the keys in the command
1672 # CLIENT TRACKING is a special case.
1673 # It doesn't have any keys, it needs to be sent to the provided nodes
1674 # By default it will be sent to all nodes.
1675 if command.upper() == "CLIENT TRACKING":
1676 return None
1678 # EVAL and EVALSHA are common enough that it's wasteful to go to the
1679 # redis server to parse the keys. Besides, there is a bug in redis<7.0
1680 # where `self._get_command_keys()` fails anyway. So, we special case
1681 # EVAL/EVALSHA.
1682 if command.upper() in ("EVAL", "EVALSHA"):
1683 # command syntax: EVAL "script body" num_keys ...
1684 if len(args) <= 2:
1685 raise RedisClusterException(f"Invalid args in command: {args}")
1686 num_actual_keys = int(args[2])
1687 eval_keys = args[3 : 3 + num_actual_keys]
1688 # if there are 0 keys, that means the script can be run on any node
1689 # so we can just return a random slot
1690 if len(eval_keys) == 0:
1691 return random.randrange(0, REDIS_CLUSTER_HASH_SLOTS)
1692 keys = eval_keys
1693 else:
1694 keys = self._get_command_keys(*args)
1695 if keys is None or len(keys) == 0:
1696 # FCALL can call a function with 0 keys, that means the function
1697 # can be run on any node so we can just return a random slot
1698 if command.upper() in ("FCALL", "FCALL_RO"):
1699 return random.randrange(0, REDIS_CLUSTER_HASH_SLOTS)
1700 raise RedisClusterException(
1701 "No way to dispatch this command to Redis Cluster. "
1702 "Missing key.\nYou can execute the command by specifying "
1703 f"target nodes.\nCommand: {args}"
1704 )
1706 # single key command
1707 if len(keys) == 1:
1708 return self.keyslot(keys[0])
1710 # multi-key command; we need to make sure all keys are mapped to
1711 # the same slot
1712 slots = {self.keyslot(key) for key in keys}
1713 if len(slots) != 1:
1714 raise RedisClusterException(
1715 f"{command} - all keys must map to the same key slot"
1716 )
1718 return slots.pop()
1720 def get_encoder(self):
1721 """
1722 Get the connections' encoder
1723 """
1724 return self.encoder
1726 def get_connection_kwargs(self):
1727 """
1728 Get the connections' key-word arguments
1729 """
1730 return self.nodes_manager.connection_kwargs
1732 def _is_nodes_flag(self, target_nodes):
1733 return isinstance(target_nodes, str) and target_nodes in self.node_flags
1735 def _parse_target_nodes(self, target_nodes):
1736 if isinstance(target_nodes, list):
1737 nodes = target_nodes
1738 elif isinstance(target_nodes, ClusterNode):
1739 # Supports passing a single ClusterNode as a variable
1740 nodes = [target_nodes]
1741 elif isinstance(target_nodes, dict):
1742 # Supports dictionaries of the format {node_name: node}.
1743 # It enables to execute commands with multi nodes as follows:
1744 # rc.cluster_save_config(rc.get_primaries())
1745 nodes = target_nodes.values()
1746 else:
1747 raise TypeError(
1748 "target_nodes type can be one of the following: "
1749 "node_flag (PRIMARIES, REPLICAS, RANDOM, ALL_NODES),"
1750 "ClusterNode, list<ClusterNode>, or dict<any, ClusterNode>. "
1751 f"The passed type is {type(target_nodes)}"
1752 )
1753 return nodes
1755 def execute_command(self, *args, **kwargs):
1756 return self._internal_execute_command(*args, **kwargs)
1758 def _internal_execute_command(self, *args, **kwargs):
1759 """
1760 Wrapper for ERRORS_ALLOW_RETRY error handling.
1762 It will try the number of times specified by the retries property from
1763 config option "self.retry" which defaults to 10 unless manually
1764 configured.
1766 If it reaches the number of times, the command will raise the exception
1768 Key argument :target_nodes: can be passed with the following types:
1769 nodes_flag: PRIMARIES, REPLICAS, ALL_NODES, RANDOM
1770 ClusterNode
1771 list<ClusterNode>
1772 dict<Any, ClusterNode>
1773 """
1774 target_nodes_specified = False
1775 is_default_node = False
1776 target_nodes = None
1777 passed_targets = kwargs.pop("target_nodes", None)
1778 if (
1779 passed_targets is not None
1780 and not self._is_nodes_flag(passed_targets)
1781 and not (
1782 isinstance(passed_targets, (list, dict, str)) and not passed_targets
1783 )
1784 ):
1785 target_nodes = self._parse_target_nodes(passed_targets)
1786 target_nodes_specified = True
1788 command, command_policies = self._resolve_command_policies(
1789 *args, target_nodes_specified=target_nodes_specified
1790 )
1792 # If an error that allows retrying was thrown, the nodes and slots
1793 # cache were reinitialized. We will retry executing the command with
1794 # the updated cluster setup only when the target nodes can be
1795 # determined again with the new cache tables. Therefore, when target
1796 # nodes were passed to this function, we cannot retry the command
1797 # execution since the nodes may not be valid anymore after the tables
1798 # were reinitialized. So in case of passed target nodes,
1799 # retry_attempts will be set to 0.
1800 retry_attempts = 0 if target_nodes_specified else self.retry.get_retries()
1801 # Add one for the first execution
1802 execute_attempts = 1 + retry_attempts
1803 failure_count = 0
1805 # Start timing for observability
1806 start_time = time.monotonic()
1808 for _ in range(execute_attempts):
1809 try:
1810 res = {}
1811 if not target_nodes_specified:
1812 # Determine the nodes to execute the command on
1813 target_nodes = self._determine_nodes(
1814 *args,
1815 request_policy=command_policies.request_policy,
1816 nodes_flag=passed_targets,
1817 )
1819 if not target_nodes:
1820 raise RedisClusterException(
1821 f"No targets were found to execute {args} command on"
1822 )
1823 if (
1824 len(target_nodes) == 1
1825 and target_nodes[0] == self.get_default_node()
1826 ):
1827 is_default_node = True
1828 for node in target_nodes:
1829 res[node.name] = self._execute_command(node, *args, **kwargs)
1831 if command_policies.response_policy == ResponsePolicy.ONE_SUCCEEDED:
1832 break
1834 # Return the processed result
1835 # ``command``, not ``args[0]``: the result callbacks are keyed by the name
1836 # the policies were decided by, so a container command passed as two words
1837 # - ``execute_command("command", "count")`` - is dispatched as the command
1838 # that was actually routed. Telemetry deliberately does not follow; see the
1839 # note on the metric in the retry branch below.
1840 return self._process_result(
1841 command,
1842 res,
1843 response_policy=command_policies.response_policy,
1844 **kwargs,
1845 )
1846 except Exception as e:
1847 if retry_attempts > 0 and type(e) in self.__class__.ERRORS_ALLOW_RETRY:
1848 if is_default_node:
1849 # Replace the default cluster node
1850 self.replace_default_node()
1851 # The nodes and slots cache were reinitialized.
1852 # Try again with the new cluster setup.
1853 retry_attempts -= 1
1854 failure_count += 1
1856 if hasattr(e, "connection"):
1857 # ``args[0]``, not the resolved ``command``: every other metric in
1858 # this class - including the per-command one ``_execute_command``
1859 # records on success - names the command the caller spelled. Using
1860 # the routed name only here would report one command under two
1861 # names depending on whether it was retried.
1862 self._record_command_metric(
1863 command_name=args[0],
1864 duration_seconds=time.monotonic() - start_time,
1865 connection=e.connection,
1866 error=e,
1867 )
1869 self._record_error_metric(
1870 error=e,
1871 connection=e.connection,
1872 retry_attempts=failure_count,
1873 )
1874 continue
1875 else:
1876 # raise the exception
1877 if hasattr(e, "connection"):
1878 self._record_error_metric(
1879 error=e,
1880 connection=e.connection,
1881 retry_attempts=failure_count,
1882 is_internal=False,
1883 )
1884 raise e
1886 def _himport_reconcile_discards(self, redis_node, connection):
1887 """Delegate to the shared sync HIMPORT executor."""
1888 return _himport_exec.reconcile_discards(redis_node, connection)
1890 def _himport_prepare_and_set(
1891 self,
1892 redis_node,
1893 connection,
1894 key,
1895 fieldset_name,
1896 values,
1897 fieldset,
1898 asking: bool = False,
1899 ):
1900 """Delegate to the shared sync HIMPORT executor."""
1901 return _himport_exec.prepare_and_set(
1902 redis_node,
1903 connection,
1904 key,
1905 fieldset_name,
1906 values,
1907 fieldset,
1908 asking=asking,
1909 )
1911 def _himport_execute_set(
1912 self,
1913 redis_node,
1914 connection,
1915 key,
1916 fieldset_name,
1917 values,
1918 asking: bool = False,
1919 ):
1920 """Delegate to the shared sync HIMPORT executor."""
1921 return _himport_exec.execute_set(
1922 redis_node, connection, key, fieldset_name, values, asking=asking
1923 )
1925 def _execute_command(self, target_node, *args, **kwargs):
1926 """
1927 Send a command to a node in the cluster
1928 """
1929 command = args[0]
1930 redis_node = None
1931 connection = None
1932 redirect_addr = None
1933 asking = False
1934 moved = False
1935 ttl = int(self.RedisClusterRequestTTL)
1937 # Start timing for observability
1938 start_time = time.monotonic()
1940 while ttl > 0:
1941 ttl -= 1
1942 try:
1943 if asking:
1944 target_node = self.get_node(node_name=redirect_addr)
1945 elif moved:
1946 # MOVED occurred and the slots cache was updated,
1947 # refresh the target node
1948 slot = self.determine_slot(*args)
1949 replica_safe = (
1950 self.read_from_replicas
1951 or self.load_balancing_strategy is not None
1952 ) and self._is_replica_safe(command)
1953 target_node = self.nodes_manager.get_node_from_slot(
1954 slot,
1955 replica_safe,
1956 self.load_balancing_strategy if replica_safe else None,
1957 )
1958 moved = False
1960 redis_node = self.get_redis_connection(target_node)
1961 connection = get_connection(redis_node)
1962 himport_set = parse_himport_set_args(args)
1963 if asking and himport_set is None:
1964 connection.send_command("ASKING")
1965 redis_node.parse_response(connection, "ASKING", **kwargs)
1966 asking = False
1967 if himport_set is not None:
1968 # args == (HIMPORT_SET, key, fieldset_name, *values). A raw
1969 # ``execute_command`` with too few args falls through to the
1970 # normal send path below so the server returns its arity error
1971 # instead of a client-side IndexError.
1972 # The cluster
1973 # executor lazily PREPAREs the fieldset on this connection and
1974 # reconciles deferred DISCARDs, then SETs; it already applies the
1975 # HIMPORT SET response callback, so it bypasses the cluster callback
1976 # block below.
1977 # This per-command branch in the hot dispatch path is deliberate
1978 # and has no cleaner alternative: this is the only seam where the
1979 # concrete routed connection is known, and connection-scoped
1980 # session setup can only happen once that connection is chosen.
1981 # On an ASK redirect ``asking`` is folded into the SET's own packed
1982 # write (see the guard above that suppresses the standalone ASKING
1983 # for HIMPORT SET) so the allowance sits immediately before the SET.
1984 # Clear ``asking`` first and carry the allowance in a dedicated
1985 # local: ``_himport_execute_set`` can raise a retriable MOVED/TRYAGAIN
1986 # mid-exchange, and a stale ``asking`` would shadow the moved-retry
1987 # branch on the next loop iteration (mirrors the async client).
1988 key, fieldset_name, values = himport_set
1989 ask_himport = asking
1990 asking = False
1991 response = self._himport_execute_set(
1992 redis_node,
1993 connection,
1994 key,
1995 fieldset_name,
1996 values,
1997 asking=ask_himport,
1998 )
1999 kwargs.pop("keys", None)
2000 else:
2001 connection.send_command(*args, **kwargs)
2002 response = redis_node.parse_response(connection, command, **kwargs)
2004 # Remove keys entry, it needs only for cache.
2005 kwargs.pop("keys", None)
2007 if command in self.cluster_response_callbacks:
2008 response = self.cluster_response_callbacks[command](
2009 response, **kwargs
2010 )
2012 self._record_command_metric(
2013 command_name=command,
2014 duration_seconds=time.monotonic() - start_time,
2015 connection=connection,
2016 )
2017 return response
2018 except AuthenticationError as e:
2019 e.connection = connection if connection is not None else target_node
2020 self._record_command_metric(
2021 command_name=command,
2022 duration_seconds=time.monotonic() - start_time,
2023 connection=e.connection,
2024 error=e,
2025 )
2026 raise
2027 except MaxConnectionsError as e:
2028 # MaxConnectionsError indicates client-side resource exhaustion
2029 # (too many connections in the pool), not a node failure.
2030 # Don't treat this as a node failure - just re-raise the error
2031 # without reinitializing the cluster.
2032 # The connection in the error is used to report the metrics based on host and port info
2033 # so we use the target node object which contains the host and port info
2034 # because we did not get the connection yet
2035 e.connection = target_node
2036 self._record_command_metric(
2037 command_name=command,
2038 duration_seconds=time.monotonic() - start_time,
2039 connection=e.connection,
2040 error=e,
2041 )
2042 raise
2043 except (ConnectionError, TimeoutError) as e:
2044 if is_debug_log_enabled():
2045 connection_details = (
2046 connection.extract_connection_details()
2047 if connection
2048 else "no connection"
2049 )
2050 # Log only the command name - argument values can carry
2051 # secrets or user data.
2052 args_log_str = safe_str(args[0])
2053 logger.debug(
2054 f"{type(e).__name__} received for command {args_log_str}, on node {target_node.name}, "
2055 f"and connection: {connection}, {connection_details}, error: {e}"
2056 )
2057 # this is used to report the metrics based on host and port info
2058 e.connection = connection if connection else target_node
2060 # ConnectionError can also be raised if we couldn't get a
2061 # connection from the pool before timing out, so check that
2062 # this is an actual connection before attempting to disconnect.
2063 if connection is not None:
2064 connection.disconnect()
2066 # Instead of setting to None, properly handle the pool
2067 # Get the pool safely - redis_connection could be set to None
2068 # by another thread between the check and access
2069 redis_conn = target_node.redis_connection
2070 if redis_conn is not None:
2071 pool = redis_conn.connection_pool
2072 if pool is not None:
2073 with pool._lock:
2074 # take care for the active connections in the pool
2075 pool.update_active_connections_for_reconnect()
2076 # disconnect all free connections
2077 pool.disconnect_free_connections()
2079 # Move the failed node to the end of the cached nodes list
2080 self.nodes_manager.move_node_to_end_of_cached_nodes(target_node.name)
2082 # DON'T set redis_connection = None - keep the pool for reuse
2083 # provide the name of the failed node so we can try it last
2084 self.nodes_manager.initialize(last_failed_node_name=target_node.name)
2085 self._record_command_metric(
2086 command_name=command,
2087 duration_seconds=time.monotonic() - start_time,
2088 connection=e.connection,
2089 error=e,
2090 )
2091 raise e
2092 except MovedError as e:
2093 if is_debug_log_enabled():
2094 connection_details = (
2095 connection.extract_connection_details()
2096 if connection
2097 else "no connection"
2098 )
2099 # Log only the command name - argument values can carry
2100 # secrets or user data.
2101 args_log_str = safe_str(args[0])
2102 logger.debug(
2103 f"MOVED error received for command {args_log_str}, on node {target_node.name}, "
2104 f"and connection: {connection}, {connection_details}, error: {e}"
2105 )
2106 # First, we will try to patch the slots/nodes cache with the
2107 # redirected node output and try again. If MovedError exceeds
2108 # 'reinitialize_steps' number of times, we will force
2109 # reinitializing the tables, and then try again.
2110 # 'reinitialize_steps' counter will increase faster when
2111 # the same client object is shared between multiple threads. To
2112 # reduce the frequency you can set this variable in the
2113 # RedisCluster constructor.
2114 self.reinitialize_counter += 1
2115 if self._should_reinitialized():
2116 # during this call all connections are closed or marked for disconnect,
2117 # so we don't need to disconnect the changed node's connections
2118 self.nodes_manager.initialize(
2119 additional_startup_nodes_info=[(e.host, e.port)]
2120 )
2121 # Reset the counter
2122 self.reinitialize_counter = 0
2123 else:
2124 self.nodes_manager.move_slot(e)
2125 moved = True
2126 self._record_command_metric(
2127 command_name=command,
2128 duration_seconds=time.monotonic() - start_time,
2129 connection=connection,
2130 error=e,
2131 )
2132 self._record_error_metric(
2133 error=e,
2134 connection=connection,
2135 )
2136 except TryAgainError as e:
2137 if is_debug_log_enabled():
2138 connection_details = (
2139 connection.extract_connection_details()
2140 if connection
2141 else "no connection"
2142 )
2143 # Log only the command name - argument values can carry
2144 # secrets or user data.
2145 args_log_str = safe_str(args[0])
2146 logger.debug(
2147 f"TRYAGAIN error received for command {args_log_str}, on node {target_node.name}, "
2148 f"and connection: {connection}, {connection_details}"
2149 )
2150 if ttl < self.RedisClusterRequestTTL / 2:
2151 time.sleep(0.05)
2153 self._record_command_metric(
2154 command_name=command,
2155 duration_seconds=time.monotonic() - start_time,
2156 connection=connection,
2157 error=e,
2158 )
2159 self._record_error_metric(
2160 error=e,
2161 connection=connection,
2162 )
2163 except AskError as e:
2164 if is_debug_log_enabled():
2165 connection_details = (
2166 connection.extract_connection_details()
2167 if connection
2168 else "no connection"
2169 )
2170 # Log only the command name - argument values can carry
2171 # secrets or user data.
2172 args_log_str = safe_str(args[0])
2173 logger.debug(
2174 f"ASK error received for command {args_log_str}, on node {target_node.name}, "
2175 f"and connection: {connection}, {connection_details}, error: {e}"
2176 )
2177 redirect_addr = get_node_name(host=e.host, port=e.port)
2178 asking = True
2180 self._record_command_metric(
2181 command_name=command,
2182 duration_seconds=time.monotonic() - start_time,
2183 connection=connection,
2184 error=e,
2185 )
2186 self._record_error_metric(
2187 error=e,
2188 connection=connection,
2189 )
2190 except (ClusterDownError, SlotNotCoveredError) as e:
2191 # ClusterDownError can occur during a failover and to get
2192 # self-healed, we will try to reinitialize the cluster layout
2193 # and retry executing the command
2195 # SlotNotCoveredError can occur when the cluster is not fully
2196 # initialized or can be temporary issue.
2197 # We will try to reinitialize the cluster topology
2198 # and retry executing the command
2200 time.sleep(0.25)
2201 self.nodes_manager.initialize()
2203 # if we have a connection, use it, otherwise use the target node
2204 # object which contains the host and port info
2205 # this is used to report the metrics based on host and port info
2206 e.connection = connection if connection else target_node
2207 self._record_command_metric(
2208 command_name=command,
2209 duration_seconds=time.monotonic() - start_time,
2210 connection=e.connection,
2211 error=e,
2212 )
2213 raise
2214 except ResponseError as e:
2215 # this is used to report the metrics based on host and port info
2216 # ResponseError typically happens after get_connection() succeeds,
2217 # so connection should be available
2218 e.connection = connection if connection else target_node
2219 self._record_command_metric(
2220 command_name=command,
2221 duration_seconds=time.monotonic() - start_time,
2222 connection=e.connection,
2223 error=e,
2224 )
2225 raise
2226 except Exception as e:
2227 if connection:
2228 connection.disconnect()
2230 # if we have a connection, use it, otherwise use the target node
2231 # object which contains the host and port info
2232 # this is used to report the metrics based on host and port info
2233 e.connection = connection if connection else target_node
2234 self._record_command_metric(
2235 command_name=command,
2236 duration_seconds=time.monotonic() - start_time,
2237 connection=e.connection,
2238 error=e,
2239 )
2240 raise e
2241 finally:
2242 if connection is not None:
2243 redis_node.connection_pool.release(connection)
2245 e = ClusterError("TTL exhausted.")
2246 # In this case we should have an active connection.
2247 # If we are here, we have received many MOVED or ASK errors and finally exhausted the TTL.
2248 # This means that we used an active connection to read from the socket.
2249 # This is used to report metrics based on the host and port information.
2250 e.connection = connection
2251 self._record_command_metric(
2252 command_name=command,
2253 duration_seconds=time.monotonic() - start_time,
2254 connection=connection,
2255 error=e,
2256 )
2257 raise e
2259 def _record_command_metric(
2260 self,
2261 command_name: str,
2262 duration_seconds: float,
2263 connection: Connection,
2264 error=None,
2265 ):
2266 """
2267 Records operation duration metric directly.
2268 """
2269 host = connection.host if connection else "unknown"
2270 port = connection.port if connection else 0
2271 db = str(connection.db) if connection and hasattr(connection, "db") else "0"
2273 record_operation_duration(
2274 command_name=command_name,
2275 duration_seconds=duration_seconds,
2276 server_address=host,
2277 server_port=port,
2278 db_namespace=db,
2279 error=error,
2280 )
2282 def _record_error_metric(
2283 self,
2284 error: Exception,
2285 connection: Connection,
2286 is_internal: bool = True,
2287 retry_attempts: Optional[int] = None,
2288 ):
2289 """
2290 Records error count metric directly.
2291 """
2292 record_error_count(
2293 server_address=connection.host,
2294 server_port=connection.port,
2295 network_peer_address=connection.host,
2296 network_peer_port=connection.port,
2297 error_type=error,
2298 retry_attempts=retry_attempts if retry_attempts is not None else 0,
2299 is_internal=is_internal,
2300 )
2302 def close(self) -> None:
2303 try:
2304 with self._lock:
2305 if self.nodes_manager:
2306 self.nodes_manager.close()
2307 except AttributeError:
2308 # RedisCluster's __init__ can fail before nodes_manager is set
2309 pass
2311 def _process_result(self, command, res, response_policy: ResponsePolicy, **kwargs):
2312 """
2313 Process the result of the executed command.
2314 The function would return a dict or a single value.
2316 :type command: str
2317 :type res: dict
2319 `res` should be in the following format:
2320 Dict<node_name, command_result>
2321 """
2322 if command in self.result_callbacks:
2323 res = self.result_callbacks[command](command, res, **kwargs)
2324 elif len(res) == 1:
2325 # When we execute the command on a single node, we can
2326 # remove the dictionary and return a single response
2327 res = list(res.values())[0]
2329 return self._policies_callback_mapping[response_policy](res)
2331 def load_external_module(self, funcname, func):
2332 """
2333 This function can be used to add externally defined redis modules,
2334 and their namespaces to the redis client.
2336 ``funcname`` - A string containing the name of the function to create
2337 ``func`` - The function, being added to this class.
2338 """
2339 setattr(self, funcname, func)
2341 def transaction(self, func, *watches, **kwargs):
2342 """
2343 Convenience method for executing the callable `func` as a transaction
2344 while watching all keys specified in `watches`. The 'func' callable
2345 should expect a single argument which is a Pipeline object.
2346 """
2347 shard_hint = kwargs.pop("shard_hint", None)
2348 value_from_callable = kwargs.pop("value_from_callable", False)
2349 watch_delay = kwargs.pop("watch_delay", None)
2350 with self.pipeline(True, shard_hint) as pipe:
2351 while True:
2352 try:
2353 if watches:
2354 pipe.watch(*watches)
2355 func_value = func(pipe)
2356 exec_value = pipe.execute()
2357 return func_value if value_from_callable else exec_value
2358 except WatchError:
2359 if watch_delay is not None and watch_delay > 0:
2360 time.sleep(watch_delay)
2361 continue
2364class ClusterNode:
2365 def __init__(self, host, port, server_type=None, redis_connection=None):
2366 if host == "localhost":
2367 host = socket.gethostbyname(host)
2369 self.host = host
2370 self.port = port
2371 self.name = get_node_name(host, port)
2372 self.server_type = server_type
2373 self.redis_connection = redis_connection
2375 def __repr__(self):
2376 return (
2377 f"[host={self.host},"
2378 f"port={self.port},"
2379 f"name={self.name},"
2380 f"server_type={self.server_type},"
2381 f"redis_connection={self.redis_connection}]"
2382 )
2384 def __eq__(self, obj):
2385 return isinstance(obj, ClusterNode) and obj.name == self.name
2387 def __hash__(self):
2388 return hash(self.name)
2391class LoadBalancingStrategy(Enum):
2392 ROUND_ROBIN = "round_robin"
2393 ROUND_ROBIN_REPLICAS = "round_robin_replicas"
2394 RANDOM = "random"
2395 RANDOM_REPLICA = "random_replica"
2398# The strategies that exclude the primary. ``LoadBalancer`` derives the same thing per
2399# strategy as the ``replicas_only`` bit it passes to its index helpers below, but it can
2400# only express it as an index into one shard's node list. Named here so the keyless
2401# routing path - which has no shard, and so no list to index - can ask the same question.
2402# A strategy added to the enum has to be classified here too.
2403_REPLICAS_ONLY_STRATEGIES = frozenset(
2404 {
2405 LoadBalancingStrategy.ROUND_ROBIN_REPLICAS,
2406 LoadBalancingStrategy.RANDOM_REPLICA,
2407 }
2408)
2411class LoadBalancer:
2412 """
2413 Round-Robin Load Balancing
2414 """
2416 def __init__(self, start_index: int = 0) -> None:
2417 self.primary_to_idx: dict[str, int] = {}
2418 self.start_index: int = start_index
2419 self._lock: threading.Lock = threading.Lock()
2421 def get_server_index(
2422 self,
2423 primary: str,
2424 list_size: int,
2425 load_balancing_strategy: LoadBalancingStrategy = LoadBalancingStrategy.ROUND_ROBIN,
2426 ) -> int:
2427 if load_balancing_strategy == LoadBalancingStrategy.RANDOM_REPLICA:
2428 return self._get_random_server_index(
2429 list_size,
2430 replicas_only=True,
2431 )
2432 elif load_balancing_strategy == LoadBalancingStrategy.RANDOM:
2433 return self._get_random_server_index(
2434 list_size,
2435 replicas_only=False,
2436 )
2437 else:
2438 return self._get_round_robin_index(
2439 primary,
2440 list_size,
2441 load_balancing_strategy == LoadBalancingStrategy.ROUND_ROBIN_REPLICAS,
2442 )
2444 def reset(self) -> None:
2445 with self._lock:
2446 self.primary_to_idx.clear()
2448 def _get_random_server_index(self, list_size: int, replicas_only: bool) -> int:
2449 return random.randint(1 if replicas_only else 0, list_size - 1)
2451 def _get_round_robin_index(
2452 self, primary: str, list_size: int, replicas_only: bool
2453 ) -> int:
2454 with self._lock:
2455 server_index = self.primary_to_idx.setdefault(primary, self.start_index)
2456 if replicas_only and server_index == 0:
2457 # skip the primary node index
2458 server_index = 1
2459 # Update the index for the next round
2460 self.primary_to_idx[primary] = (server_index + 1) % list_size
2461 return server_index
2464class NodesManager:
2465 def __init__(
2466 self,
2467 startup_nodes: list[ClusterNode],
2468 from_url=False,
2469 require_full_coverage=False,
2470 lock: Optional[threading.RLock] = None,
2471 dynamic_startup_nodes=True,
2472 connection_pool_class=ConnectionPool,
2473 address_remap: Optional[Callable[[Tuple[str, int]], Tuple[str, int]]] = None,
2474 cache: Optional[CacheInterface] = None,
2475 cache_config: Optional[CacheConfig] = None,
2476 cache_factory: Optional[CacheFactoryInterface] = None,
2477 event_dispatcher: Optional[EventDispatcher] = None,
2478 maint_notifications_config: Optional[MaintNotificationsConfig] = None,
2479 himport_registry: HImportRegistry | None = None,
2480 metadata_resolver: Optional[MetadataResolver] = None,
2481 **kwargs,
2482 ):
2483 # Shared, cluster-wide HIMPORT registry object, injected onto every node's pool
2484 # in create_redis_node (not forwarded through connection_kwargs, so all nodes
2485 # reuse the one object rather than rebuilding it per node).
2486 self.himport_registry = himport_registry
2487 self.nodes_cache: dict[str, ClusterNode] = {}
2488 self.slots_cache: dict[int, list[ClusterNode]] = {}
2489 self.startup_nodes: dict[str, ClusterNode] = {n.name: n for n in startup_nodes}
2490 self.default_node: Optional[ClusterNode] = None
2491 self._epoch: int = 0
2492 self.from_url = from_url
2493 self._require_full_coverage = require_full_coverage
2494 self._dynamic_startup_nodes = dynamic_startup_nodes
2495 self.connection_pool_class = connection_pool_class
2496 self.address_remap = address_remap
2497 # Shared, cluster-wide metadata resolver, injected onto every node's client in
2498 # create_redis_node for the same reason the cache and the HIMPORT registry are:
2499 # every node must resolve command metadata - and therefore cache eligibility -
2500 # through the one object the cluster client was configured with.
2501 self._metadata_resolver = metadata_resolver
2503 self._cache: Optional[CacheInterface] = None
2504 if cache:
2505 self._cache = cache
2506 elif cache_factory is not None:
2507 self._cache = cache_factory.get_cache()
2508 elif cache_config is not None:
2509 # Injected here, on a copy, rather than left to the node pools: the cluster hands
2510 # every node the one cache built below, so each pool sees a ``cache=`` and would
2511 # set the resolver on the configuration inside it - which is the caller's object,
2512 # since ``CacheFactory`` holds it by reference. Copying keeps a ``CacheConfig``
2513 # reused across clients from picking up whichever resolver was injected last,
2514 # exactly as ``ConnectionPool.__init__`` does for the standalone client.
2515 #
2516 # Only this branch needs it. ``cache=`` and ``cache_factory=`` hand over a whole
2517 # cache whose configuration the caller owns, and the node pools set the resolver
2518 # on it in place - the same thing they do for a standalone client given one.
2519 if metadata_resolver is not None and isinstance(cache_config, CacheConfig):
2520 cache_config = copy(cache_config)
2521 cache_config.set_metadata_resolver(metadata_resolver)
2523 self._cache = CacheFactory(cache_config).get_cache()
2524 self.connection_kwargs = kwargs
2525 self.read_load_balancer = LoadBalancer()
2527 # nodes_cache / slots_cache / startup_nodes / default_node are protected by _lock
2528 if lock is None:
2529 self._lock = threading.RLock()
2530 else:
2531 self._lock = lock
2533 # initialize holds _initialization_lock to dedup multiple calls to reinitialize;
2534 # note that if we hold both _lock and _initialization_lock, we _must_ acquire
2535 # _initialization_lock first (ie: to have a consistent order) to avoid deadlock.
2536 #
2537 # The same ordering rule extends to OSSMaintNotificationsHandler._lock, which
2538 # is a third lock in this graph: initialize runs a CLUSTER SLOTS round trip
2539 # while holding _initialization_lock, and the response can carry an SMIGRATED
2540 # push that is handled inline on that thread and needs the handler's _lock. The
2541 # full order is therefore
2542 # _initialization_lock -> OSSMaintNotificationsHandler._lock
2543 # -> NodesManager._lock / connection pool locks
2544 # ie: a thread holding the handler's _lock must never wait for
2545 # _initialization_lock.
2546 self._initialization_lock: threading.RLock = threading.RLock()
2547 # Ident of the thread currently running initialize, or None. Written only
2548 # under _initialization_lock; see the re-entrancy guard in initialize.
2549 self._initializing_thread_id: Optional[int] = None
2551 if event_dispatcher is None:
2552 self._event_dispatcher = EventDispatcher()
2553 else:
2554 self._event_dispatcher = event_dispatcher
2555 self._credential_provider = self.connection_kwargs.get(
2556 "credential_provider", None
2557 )
2558 self.maint_notifications_config = maint_notifications_config
2560 self.initialize()
2562 def get_node(
2563 self,
2564 host: Optional[str] = None,
2565 port: Optional[int] = None,
2566 node_name: Optional[str] = None,
2567 ) -> Optional[ClusterNode]:
2568 """
2569 Get the requested node from the cluster's nodes.
2570 nodes.
2571 :return: ClusterNode if the node exists, else None
2572 """
2573 if host and port:
2574 # the user passed host and port
2575 if host == "localhost":
2576 host = socket.gethostbyname(host)
2577 with self._lock:
2578 return self.nodes_cache.get(get_node_name(host=host, port=port))
2579 elif node_name:
2580 with self._lock:
2581 return self.nodes_cache.get(node_name)
2582 else:
2583 return None
2585 def move_slot(self, e: Union[AskError, MovedError]):
2586 """
2587 Update the slot's node with the redirected one
2588 """
2589 node_changed = False
2590 with self._lock:
2591 redirected_node = self.get_node(host=e.host, port=e.port)
2592 if redirected_node is not None:
2593 # The node already exists
2594 if redirected_node.server_type is not PRIMARY:
2595 # Update the node's server type
2596 redirected_node.server_type = PRIMARY
2597 else:
2598 # This is a new node, we will add it to the nodes cache
2599 redirected_node = ClusterNode(e.host, e.port, PRIMARY)
2600 self.nodes_cache[redirected_node.name] = redirected_node
2602 slot_nodes = self.slots_cache[e.slot_id]
2603 if redirected_node not in slot_nodes:
2604 # The new slot owner is a new server, or a server from a different
2605 # shard. We need to remove all current nodes from the slot's list
2606 # (including replications) and add just the new node.
2607 self.slots_cache[e.slot_id] = [redirected_node]
2608 node_changed = True
2609 elif redirected_node is not slot_nodes[0]:
2610 # The MOVED error resulted from a failover, and the new slot owner
2611 # had previously been a replica.
2612 old_primary = slot_nodes[0]
2613 # Update the old primary to be a replica and add it to the end of
2614 # the slot's node list
2615 old_primary.server_type = REPLICA
2616 slot_nodes.append(old_primary)
2617 # Remove the old replica, which is now a primary, from the slot's
2618 # node list
2619 slot_nodes.remove(redirected_node)
2620 # Override the old primary with the new one
2621 slot_nodes[0] = redirected_node
2622 if self.default_node == old_primary:
2623 # Update the default node with the new primary
2624 self.default_node = redirected_node
2625 node_changed = True
2626 # else: circular MOVED to current primary -> no-op
2627 # Dispatch outside the lock so listeners can acquire their own locks
2628 # without risk of deadlock. Skipped on the no-op branch to avoid
2629 # needless reconciliation walks under MOVED storms. A listener must
2630 # not break slots-cache refresh; log and continue so a single buggy
2631 # listener cannot starve the rest.
2632 if node_changed:
2633 try:
2634 self._event_dispatcher.dispatch(AfterSlotsCacheRefreshEvent())
2635 except Exception as exc:
2636 # Don't shadow the method parameter ``e``: ``except as`` binds
2637 # the listener exception in the function scope and ``del``s
2638 # the name on block exit (PEP 3134), which would also wipe
2639 # out the original AskError/MovedError parameter.
2640 logger.exception(
2641 "listener raised during slots-cache refresh: %s: %s",
2642 type(exc).__name__,
2643 exc,
2644 )
2646 @deprecated_args(
2647 args_to_warn=["server_type"],
2648 reason=(
2649 "In case you need select some load balancing strategy "
2650 "that will use replicas, please set it through 'load_balancing_strategy'"
2651 ),
2652 version="5.3.0",
2653 )
2654 def get_node_from_slot(
2655 self,
2656 slot: int,
2657 read_from_replicas: bool = False,
2658 load_balancing_strategy: Optional[LoadBalancingStrategy] = None,
2659 server_type: Optional[Literal["primary", "replica"]] = None,
2660 ) -> ClusterNode:
2661 """
2662 Gets a node that servers this hash slot
2663 """
2665 if read_from_replicas is True and load_balancing_strategy is None:
2666 load_balancing_strategy = LoadBalancingStrategy.ROUND_ROBIN
2668 with self._lock:
2669 if self.slots_cache.get(slot) is None or len(self.slots_cache[slot]) == 0:
2670 raise SlotNotCoveredError(
2671 f'Slot "{slot}" not covered by the cluster. '
2672 + f'"require_full_coverage={self._require_full_coverage}"'
2673 )
2675 if len(self.slots_cache[slot]) > 1 and load_balancing_strategy:
2676 # get the server index using the strategy defined in load_balancing_strategy
2677 primary_name = self.slots_cache[slot][0].name
2678 node_idx = self.read_load_balancer.get_server_index(
2679 primary_name, len(self.slots_cache[slot]), load_balancing_strategy
2680 )
2681 elif (
2682 server_type is None
2683 or server_type == PRIMARY
2684 or len(self.slots_cache[slot]) == 1
2685 ):
2686 # return a primary
2687 node_idx = 0
2688 else:
2689 # return a replica
2690 # randomly choose one of the replicas
2691 node_idx = random.randint(1, len(self.slots_cache[slot]) - 1)
2693 return self.slots_cache[slot][node_idx]
2695 def get_nodes_by_server_type(self, server_type: Literal["primary", "replica"]):
2696 """
2697 Get all nodes with the specified server type
2698 :param server_type: 'primary' or 'replica'
2699 :return: list of ClusterNode
2700 """
2701 with self._lock:
2702 return [
2703 node
2704 for node in self.nodes_cache.values()
2705 if node.server_type == server_type
2706 ]
2708 @deprecated_function(
2709 reason="This method is not used anymore internally. The startup nodes are populated automatically.",
2710 version="7.0.2",
2711 )
2712 def populate_startup_nodes(self, nodes):
2713 """
2714 Populate all startup nodes and filters out any duplicates
2715 """
2716 with self._lock:
2717 for n in nodes:
2718 self.startup_nodes[n.name] = n
2720 def move_node_to_end_of_cached_nodes(self, node_name: str) -> None:
2721 """
2722 Move a failing node to the end of startup_nodes and nodes_cache so it's
2723 tried last during reinitialization and when selecting the default node.
2724 If the node is not in the respective list, nothing is done.
2725 """
2726 # Move in startup_nodes
2727 if node_name in self.startup_nodes and len(self.startup_nodes) > 1:
2728 node = self.startup_nodes.pop(node_name)
2729 self.startup_nodes[node_name] = node # Re-insert at end
2731 # Move in nodes_cache - this affects get_nodes_by_server_type ordering
2732 # which is used to select the default_node during initialize()
2733 if node_name in self.nodes_cache and len(self.nodes_cache) > 1:
2734 node = self.nodes_cache.pop(node_name)
2735 self.nodes_cache[node_name] = node # Re-insert at end
2737 def check_slots_coverage(self, slots_cache):
2738 # Validate if all slots are covered or if we should try next
2739 # startup node
2740 for i in range(0, REDIS_CLUSTER_HASH_SLOTS):
2741 if i not in slots_cache:
2742 return False
2743 return True
2745 def create_redis_connections(self, nodes):
2746 """
2747 This function will create a redis connection to all nodes in :nodes:
2748 """
2749 connection_pools = []
2750 for node in nodes:
2751 if node.redis_connection is None:
2752 node.redis_connection = self.create_redis_node(
2753 host=node.host,
2754 port=node.port,
2755 maint_notifications_config=self.maint_notifications_config,
2756 **self.connection_kwargs,
2757 )
2758 connection_pools.append(node.redis_connection.connection_pool)
2760 self._event_dispatcher.dispatch(
2761 AfterPooledConnectionsInstantiationEvent(
2762 connection_pools, ClientType.SYNC, self._credential_provider
2763 )
2764 )
2766 def create_redis_node(
2767 self,
2768 host,
2769 port,
2770 **kwargs,
2771 ):
2772 # We are configuring the connection pool not to retry
2773 # connections on lower level clients to avoid retrying
2774 # connections to nodes that are not reachable
2775 # and to avoid blocking the connection pool.
2776 # The only error that will have some handling in the lower
2777 # level clients is ConnectionError which will trigger disconnection
2778 # of the socket.
2779 # The retries will be handled on cluster client level
2780 # where we will have proper handling of the cluster topology
2781 node_retry_config = Retry(
2782 backoff=NoBackoff(), retries=0, supported_errors=(ConnectionError,)
2783 )
2785 if self.from_url:
2786 # Create a redis node with a custom connection pool
2787 kwargs.update({"host": host})
2788 kwargs.update({"port": port})
2789 kwargs.update({"cache": self._cache})
2790 kwargs.update({"metadata_resolver": self._metadata_resolver})
2791 kwargs.update({"retry": node_retry_config})
2792 r = Redis(connection_pool=self.connection_pool_class(**kwargs))
2793 else:
2794 r = Redis(
2795 host=host,
2796 port=port,
2797 cache=self._cache,
2798 metadata_resolver=self._metadata_resolver,
2799 retry=node_retry_config,
2800 **kwargs,
2801 )
2802 # Share the one cluster-wide HIMPORT registry with this node's pool. Injected
2803 # here (rather than forwarded via connection_kwargs) so every node reuses the
2804 # same object; the node has no connections yet, so this is safe.
2805 if self.himport_registry is not None:
2806 r.connection_pool.himport_registry = self.himport_registry
2807 r.connection_pool.connection_kwargs["himport_registry"] = (
2808 self.himport_registry
2809 )
2810 return r
2812 def _get_or_create_cluster_node(self, host, port, role, tmp_nodes_cache):
2813 node_name = get_node_name(host, port)
2814 # check if we already have this node in the tmp_nodes_cache
2815 target_node = tmp_nodes_cache.get(node_name)
2816 if target_node is None:
2817 # before creating a new cluster node, check if the cluster node already
2818 # exists in the current nodes cache and has a valid connection so we can
2819 # reuse it
2820 redis_connection: Optional[Redis] = None
2821 with self._lock:
2822 previous_node = self.nodes_cache.get(node_name)
2823 if previous_node:
2824 redis_connection = previous_node.redis_connection
2825 # don't update the old ClusterNode, so we don't update its role
2826 # outside of the lock
2827 target_node = ClusterNode(host, port, role, redis_connection)
2828 # add this node to the nodes cache
2829 tmp_nodes_cache[target_node.name] = target_node
2831 return target_node
2833 def _get_epoch(self) -> int:
2834 """
2835 Get the current epoch value. This method exists primarily to allow
2836 tests to mock the epoch fetch and control race condition timing.
2837 """
2838 with self._lock:
2839 return self._epoch
2841 @contextmanager
2842 def _initializing_on_this_thread(self):
2843 """
2844 Mark the calling thread as the one currently running ``initialize``.
2846 Must be entered while holding ``_initialization_lock`` - that is what
2847 makes the calling thread the only writer of ``_initializing_thread_id``.
2848 """
2849 self._initializing_thread_id = threading.get_ident()
2850 try:
2851 yield
2852 finally:
2853 self._initializing_thread_id = None
2855 def initialize(
2856 self,
2857 additional_startup_nodes_info: Optional[List[Tuple[str, int]]] = None,
2858 disconnect_startup_nodes_pools: bool = True,
2859 last_failed_node_name: Optional[str] = None,
2860 ):
2861 """
2862 Initializes the nodes cache, slots cache and redis connections.
2863 :startup_nodes:
2864 Responsible for discovering other nodes in the cluster
2865 :disconnect_startup_nodes_pools:
2866 Whether to disconnect the connection pool of the startup nodes
2867 after the initialization is complete. This is useful when the
2868 startup nodes are not part of the cluster and we want to avoid
2869 keeping the connection open.
2870 :additional_startup_nodes_info:
2871 Additional nodes to add temporarily to the startup nodes.
2872 The additional nodes will be used just in the process of extraction of the slots
2873 and nodes information from the cluster.
2874 This is useful when we want to add new nodes to the cluster
2875 and initialize the client
2876 with them.
2877 The format of the list is a list of tuples, where each tuple contains
2878 the host and port of the node.
2879 :last_failed_node_name:
2880 Name of the node that just failed and should be tried only after
2881 other startup and additional startup nodes during this refresh.
2882 """
2883 if self._initializing_thread_id == threading.get_ident():
2884 # Re-entrant call on the thread that is already refreshing the
2885 # topology: a push notification (SMIGRATED) arrived on the CLUSTER
2886 # SLOTS response below and was handled inline on this thread. The
2887 # outer call is mid-refresh and will publish its own result, so
2888 # running a nested refresh here would reset() and swap the caches
2889 # underneath it, only for the outer call to overwrite them again
2890 # with its older snapshot. Skip instead; the outer refresh reads the
2891 # authoritative slot map anyway, and anything it still misses is
2892 # recovered through MOVED redirection.
2893 #
2894 # Reading the attribute without the lock is safe: while this thread
2895 # holds _initialization_lock it is the only writer, so a match can
2896 # only ever mean "this thread set it". A stale ident belonging to
2897 # another thread simply fails the comparison and falls through to
2898 # the normal blocking acquire below.
2899 if is_debug_log_enabled():
2900 logger.debug(
2901 "Topology refresh: skipping re-entrant initialize on thread "
2902 f"{threading.get_ident()}"
2903 )
2904 return
2905 self.reset()
2906 tmp_nodes_cache = {}
2907 tmp_slots = {}
2908 disagreements = []
2909 startup_nodes_reachable = False
2910 fully_covered = False
2911 kwargs = self.connection_kwargs
2912 exception = None
2913 epoch = self._get_epoch()
2914 if additional_startup_nodes_info is None:
2915 additional_startup_nodes_info = []
2917 with self._initialization_lock, self._initializing_on_this_thread():
2918 with self._lock:
2919 if epoch != self._epoch:
2920 # another thread has already re-initialized the nodes; don't
2921 # bother running again
2922 return
2924 with self._lock:
2925 startup_nodes = list(self.startup_nodes.values())
2926 deferred_failed_nodes = []
2927 if last_failed_node_name is not None:
2928 for index, node in enumerate(startup_nodes):
2929 if node.name == last_failed_node_name:
2930 deferred_failed_nodes.append(startup_nodes.pop(index))
2931 break
2932 if len(startup_nodes) > 1:
2933 # Vary which startup node is queried first so clients do not
2934 # all reinitialize through the same node.
2935 random.shuffle(startup_nodes)
2937 additional_startup_nodes = [
2938 ClusterNode(host, port) for host, port in additional_startup_nodes_info
2939 ]
2940 if last_failed_node_name is not None:
2941 for index, node in enumerate(additional_startup_nodes):
2942 if node.name == last_failed_node_name:
2943 if not deferred_failed_nodes:
2944 deferred_failed_nodes.append(node)
2945 additional_startup_nodes.pop(index)
2946 break
2947 if is_debug_log_enabled():
2948 logger.debug(
2949 f"Topology refresh: using additional nodes: {[node.name for node in additional_startup_nodes]}; "
2950 f"and startup nodes: {[node.name for node in startup_nodes]}"
2951 )
2953 for startup_node in chain(
2954 startup_nodes,
2955 additional_startup_nodes,
2956 deferred_failed_nodes,
2957 ):
2958 try:
2959 if startup_node.redis_connection:
2960 r = startup_node.redis_connection
2962 else:
2963 # Create a new Redis connection
2964 if is_debug_log_enabled():
2965 socket_timeout = kwargs.get("socket_timeout", "not set")
2966 socket_connect_timeout = kwargs.get(
2967 "socket_connect_timeout", "not set"
2968 )
2969 maint_enabled = (
2970 self.maint_notifications_config.enabled
2971 if self.maint_notifications_config
2972 else False
2973 )
2974 logger.debug(
2975 "Topology refresh: Creating new Redis connection to "
2976 f"{startup_node.host}:{startup_node.port}; "
2977 f"with socket_timeout: {socket_timeout}, and "
2978 f"socket_connect_timeout: {socket_connect_timeout}, "
2979 "and maint_notifications enabled: "
2980 f"{maint_enabled}"
2981 )
2982 r = self.create_redis_node(
2983 startup_node.host,
2984 startup_node.port,
2985 maint_notifications_config=self.maint_notifications_config,
2986 **kwargs,
2987 )
2988 if startup_node in self.startup_nodes.values():
2989 self.startup_nodes[startup_node.name].redis_connection = r
2990 else:
2991 startup_node.redis_connection = r
2992 try:
2993 if is_debug_log_enabled():
2994 logger.debug(
2995 "Topology refresh: querying CLUSTER SLOTS on "
2996 f"{startup_node.name}"
2997 )
2998 # Make sure cluster mode is enabled on this node
2999 cluster_slots = str_if_bytes(r.execute_command("CLUSTER SLOTS"))
3000 if disconnect_startup_nodes_pools:
3001 with r.connection_pool._lock:
3002 # take care to clear connections before we move on
3003 # mark all active connections for reconnect - they will be
3004 # reconnected on next use, but will allow current in flight commands to complete first
3005 r.connection_pool.update_active_connections_for_reconnect()
3006 # Needed to clear READONLY state when it is no longer applicable
3007 r.connection_pool.disconnect_free_connections()
3008 except ResponseError:
3009 raise RedisClusterException(
3010 "Cluster mode is not enabled on this node"
3011 )
3012 startup_nodes_reachable = True
3013 except Exception as e:
3014 # Try the next startup node.
3015 # The exception is saved and raised only if we have no more nodes.
3016 if is_debug_log_enabled():
3017 logger.debug(
3018 "Topology refresh: CLUSTER SLOTS failed on "
3019 f"{startup_node.name}: {type(e).__name__}: {e}"
3020 )
3021 exception = e
3022 continue
3024 # CLUSTER SLOTS command results in the following output:
3025 # [[slot_section[from_slot,to_slot,master,replica1,...,replicaN]]]
3026 # where each node contains the following list: [IP, port, node_id]
3027 # Therefore, cluster_slots[0][2][0] will be the IP address of the
3028 # primary node of the first slot section.
3029 # If there's only one server in the cluster, its ``host`` is ''
3030 # Fix it to the host in startup_nodes
3031 if (
3032 len(cluster_slots) == 1
3033 and len(cluster_slots[0][2][0]) == 0
3034 and len(self.startup_nodes) == 1
3035 ):
3036 cluster_slots[0][2][0] = startup_node.host
3038 for slot in cluster_slots:
3039 primary_node = slot[2]
3040 host = str_if_bytes(primary_node[0])
3041 if host == "":
3042 host = startup_node.host
3043 port = int(primary_node[1])
3044 host, port = self.remap_host_port(host, port)
3046 nodes_for_slot = []
3048 target_node = self._get_or_create_cluster_node(
3049 host, port, PRIMARY, tmp_nodes_cache
3050 )
3051 nodes_for_slot.append(target_node)
3053 replica_nodes = slot[3:]
3054 for replica_node in replica_nodes:
3055 host = str_if_bytes(replica_node[0])
3056 port = int(replica_node[1])
3057 host, port = self.remap_host_port(host, port)
3058 target_replica_node = self._get_or_create_cluster_node(
3059 host, port, REPLICA, tmp_nodes_cache
3060 )
3061 nodes_for_slot.append(target_replica_node)
3063 for i in range(int(slot[0]), int(slot[1]) + 1):
3064 if i not in tmp_slots:
3065 tmp_slots[i] = nodes_for_slot
3066 else:
3067 # Validate that 2 nodes want to use the same slot cache
3068 # setup
3069 tmp_slot = tmp_slots[i][0]
3070 if tmp_slot.name != target_node.name:
3071 disagreements.append(
3072 f"{tmp_slot.name} vs {target_node.name} on slot: {i}"
3073 )
3075 if len(disagreements) > 5:
3076 raise RedisClusterException(
3077 f"startup_nodes could not agree on a valid "
3078 f"slots cache: {', '.join(disagreements)}"
3079 )
3081 fully_covered = self.check_slots_coverage(tmp_slots)
3082 if is_debug_log_enabled():
3083 logger.debug(
3084 f"Topology refresh: CLUSTER SLOTS from {startup_node.name} "
3085 f"reported nodes {sorted(tmp_nodes_cache)}; "
3086 f"slots fully covered: {fully_covered}"
3087 )
3088 if fully_covered:
3089 # Don't need to continue to the next startup node if all
3090 # slots are covered
3091 break
3093 if not startup_nodes_reachable:
3094 # The unreachable subtype is reserved for connectivity failures:
3095 # MultiDB registers it as retryable, so a deterministic
3096 # server/configuration error (e.g. cluster mode disabled or
3097 # invalid credentials - AuthenticationError and
3098 # AuthorizationError subclass ConnectionError but cannot be
3099 # repaired by a failover) must keep surfacing as a plain
3100 # RedisClusterException.
3101 if isinstance(
3102 exception, (ConnectionError, TimeoutError, OSError)
3103 ) and not isinstance(
3104 exception, (AuthenticationError, AuthorizationError)
3105 ):
3106 raise RedisClusterUnreachableError(
3107 f"Redis Cluster cannot be connected. Please provide at least "
3108 f"one reachable node: {str(exception)}"
3109 ) from exception
3110 raise RedisClusterException(
3111 f"Redis Cluster cannot be connected. Please provide at least "
3112 f"one reachable node: {str(exception)}"
3113 ) from exception
3115 # Create Redis connections to all nodes
3116 self.create_redis_connections(list(tmp_nodes_cache.values()))
3118 # Check if the slots are not fully covered
3119 if not fully_covered and self._require_full_coverage:
3120 # Despite the requirement that the slots be covered, there
3121 # isn't a full coverage
3122 raise RedisClusterException(
3123 f"All slots are not covered after query all startup_nodes. "
3124 f"{len(tmp_slots)} of {REDIS_CLUSTER_HASH_SLOTS} "
3125 f"covered..."
3126 )
3128 # Set the tmp variables to the real variables
3129 with self._lock:
3130 self.nodes_cache = tmp_nodes_cache
3131 self.slots_cache = tmp_slots
3132 # Set the default node
3133 self.default_node = self.get_nodes_by_server_type(PRIMARY)[0]
3134 if self._dynamic_startup_nodes:
3135 # Populate the startup nodes with all discovered nodes
3136 self.startup_nodes = tmp_nodes_cache
3137 # Increment the epoch to signal that initialization has completed
3138 self._epoch += 1
3139 # Dispatch so listeners (e.g. ClusterPubSub) can reconcile per-node
3140 # state after slot ownership may have changed. A listener must not
3141 # break slots-cache refresh; log and continue so a single buggy
3142 # listener cannot starve the rest.
3143 try:
3144 self._event_dispatcher.dispatch(AfterSlotsCacheRefreshEvent())
3145 except Exception as e:
3146 logger.exception(
3147 "listener raised during slots-cache refresh: %s: %s",
3148 type(e).__name__,
3149 e,
3150 )
3152 def close(self) -> None:
3153 with self._lock:
3154 self.default_node = None
3155 nodes = tuple(self.nodes_cache.values())
3156 for node in nodes:
3157 if node.redis_connection:
3158 node.redis_connection.close()
3160 def reset(self):
3161 try:
3162 self.read_load_balancer.reset()
3163 except TypeError:
3164 # The read_load_balancer is None, do nothing
3165 pass
3167 def remap_host_port(self, host: str, port: int) -> Tuple[str, int]:
3168 """
3169 Remap the host and port returned from the cluster to a different
3170 internal value. Useful if the client is not connecting directly
3171 to the cluster.
3172 """
3173 if self.address_remap:
3174 return self.address_remap((host, port))
3175 return host, port
3177 def find_connection_owner(self, connection: Connection) -> Optional[ClusterNode]:
3178 node_name = get_node_name(connection.host, connection.port)
3179 with self._lock:
3180 for node in tuple(self.nodes_cache.values()):
3181 if node.redis_connection:
3182 conn_args = node.redis_connection.connection_pool.connection_kwargs
3183 if node_name == get_node_name(
3184 conn_args.get("host"), conn_args.get("port")
3185 ):
3186 return node
3187 return None
3190def _unregister_slots_cache_listener(
3191 dispatcher_ref: "weakref.ref[EventDispatcher]",
3192 listener: EventListenerInterface,
3193 event_type: Type[object],
3194) -> None:
3195 # Module-level finalizer callback. Kept free of strong references to the
3196 # owning ClusterPubSub so attaching it via weakref.finalize does not
3197 # extend the pubsub's lifetime.
3198 dispatcher = dispatcher_ref()
3199 if dispatcher is not None:
3200 dispatcher.unregister_listeners({event_type: [listener]})
3203class ClusterPubSubSlotsCacheListener(EventListenerInterface):
3204 """
3205 Listener that forwards AfterSlotsCacheRefreshEvent to a ClusterPubSub.
3207 Holds a weak reference to the pubsub so it does not keep the instance
3208 alive. Deterministic cleanup of the dispatcher's strong reference to this
3209 listener is performed by a ``weakref.finalize`` attached to the owning
3210 ClusterPubSub in ``ClusterPubSub.__init__``.
3211 """
3213 def __init__(self, pubsub: "ClusterPubSub") -> None:
3214 self._pubsub_ref: "weakref.ref[ClusterPubSub]" = weakref.ref(pubsub)
3216 def listen(self, event: object) -> None:
3217 pubsub = self._pubsub_ref()
3218 if pubsub is None:
3219 # Race window between pubsub GC and the finalizer running; safe
3220 # no-op, finalizer will remove this listener shortly.
3221 return
3222 try:
3223 pubsub.on_slots_changed()
3224 except Exception as e:
3225 # Listeners must not break slots-cache refresh; log and continue so
3226 # a single buggy pubsub cannot starve the rest.
3227 logger.exception(
3228 "pubsub %r raised during slots-cache change: %s: %s",
3229 pubsub,
3230 type(e).__name__,
3231 e,
3232 )
3235# How long a per-node sharded-pubsub connection is skipped by the round robin
3236# after a failed poll. PubSub._execute reconnects and retries through the
3237# connection's own Retry, so one poll on an unreachable node can cost its whole
3238# retry budget rather than the timeout the caller asked for; a cool-off keeps
3239# the single reader from spending every pass on that node while its healthy
3240# siblings hold undelivered messages.
3241SHARD_POLL_COOL_OFF_SECONDS = 1.0
3243# How long an unbounded sharded-pubsub poll waits for a not-yet-subscribed
3244# per-node pubsub before re-checking that the pubsub is still in
3245# node_pubsub_mapping. A retired one is never resubscribed and its
3246# subscribed_event is never set again, so an uninterrupted wait would park the
3247# single reader for good and stop delivery from every healthy sibling too.
3248# Only a poll that would otherwise block indefinitely ticks at all.
3249SHARD_SUBSCRIBE_WAIT_TICK_SECONDS = 1.0
3251# How often a failed sharded-pubsub poll may trigger a slots-cache refresh.
3252# Reconciliation is otherwise purely event-driven, and a node that has left the
3253# deployment answers ECONNREFUSED rather than MOVED - so without this the reader
3254# would cool off against the departed node forever and the shard channels pinned
3255# to it would never move to their new owner. Throttled because the refresh costs
3256# a CLUSTER SLOTS round trip and a failing node fails every poll.
3257SHARD_TOPOLOGY_REPAIR_INTERVAL_SECONDS = 5.0
3260class ClusterPubSub(PubSub):
3261 """
3262 Wrapper for PubSub class.
3264 IMPORTANT: before using ClusterPubSub, read about the known limitations
3265 with pubsub in Cluster mode and learn how to workaround them:
3266 https://redis.readthedocs.io/en/stable/clustering.html#known-pubsub-limitations
3267 """
3269 def __init__(
3270 self,
3271 redis_cluster,
3272 node=None,
3273 host=None,
3274 port=None,
3275 push_handler_func=None,
3276 event_dispatcher: Optional["EventDispatcher"] = None,
3277 **kwargs,
3278 ):
3279 """
3280 When a pubsub instance is created without specifying a node, a single
3281 node will be transparently chosen for the pubsub connection on the
3282 first command execution. The node will be determined by:
3283 1. Hashing the channel name in the request to find its keyslot
3284 2. Selecting a node that handles the keyslot: If read_from_replicas is
3285 set to true or load_balancing_strategy is set, a replica can be selected.
3287 :type redis_cluster: RedisCluster
3288 :type node: ClusterNode
3289 :type host: str
3290 :type port: int
3291 """
3292 self.node = None
3293 self.set_pubsub_node(redis_cluster, node, host, port)
3294 connection_pool = (
3295 None
3296 if self.node is None
3297 else redis_cluster.get_redis_connection(self.node).connection_pool
3298 )
3299 self.cluster = redis_cluster
3300 self.node_pubsub_mapping = {}
3301 # Reverse index: shard channel (normalized) -> owning node.name. Used to
3302 # route sunsubscribe calls and reconcile subscriptions after slot
3303 # migration / failover.
3304 self._shard_channel_to_node: dict = {}
3305 # Per-node poll cool-off deadlines (monotonic). Weak-keyed so a
3306 # per-node pubsub dropped from node_pubsub_mapping takes its entry with
3307 # it instead of leaking one per migration.
3308 self._poll_cool_off: "weakref.WeakKeyDictionary[PubSub, float]" = (
3309 weakref.WeakKeyDictionary()
3310 )
3311 # Node names whose last poll failed to connect. Read by
3312 # _migrate_shard_channel to skip a wire SUNSUBSCRIBE that cannot
3313 # succeed, and cleared as soon as a poll on that node works again.
3314 self._unreachable_nodes: Set[str] = set()
3315 # Monotonic deadline before which a failed poll must not trigger
3316 # another slots-cache refresh. 0.0 means "never refreshed".
3317 self._next_topology_repair: float = 0.0
3318 # Dedicated lock for shard-subscription bookkeeping. Distinct from
3319 # PubSub.self._lock (which serializes wire I/O on the cluster-level
3320 # connection used by aclose / send_command / regular subscribe) so
3321 # that reconciliation cannot starve those unrelated paths during
3322 # long per-channel migrations.
3323 self._shard_state_lock: threading.RLock = threading.RLock()
3324 # Worker executor for off-loading slot-migration reconciliation from
3325 # the dispatch call site (mirrors async's asyncio.create_task model so
3326 # the thread that triggered MovedError / topology refresh is not
3327 # blocked on per-channel sunsubscribe / ssubscribe network I/O).
3328 #
3329 # Installed by reset(), which PubSub.__init__ calls below, and replaced
3330 # by every later reset() - so it is never None once construction has
3331 # finished and the scheduling sites need no lock to create it. That
3332 # matters: on_slots_changed runs inline on whichever thread refreshed
3333 # the topology, which can be a thread that handled an SMIGRATED push and
3334 # still holds OSSMaintNotificationsHandler._lock while a third thread
3335 # waits for that handler lock holding a per-node pubsub I/O lock the
3336 # reconciliation worker needs - so blocking there on _shard_state_lock
3337 # closes a three-thread deadlock cycle.
3338 #
3339 # Constructing an executor starts no thread; ThreadPoolExecutor spawns
3340 # its worker on the first submit(), so pubsubs that never see a slot
3341 # migration never pay for one. Declared before super().__init__()
3342 # because reset() reads it.
3343 self._reconcile_executor: Optional[ThreadPoolExecutor] = None
3344 self._pubsubs_generator = self._pubsubs_generator()
3345 if event_dispatcher is None:
3346 self._event_dispatcher = EventDispatcher()
3347 else:
3348 self._event_dispatcher = event_dispatcher
3349 super().__init__(
3350 connection_pool=connection_pool,
3351 encoder=redis_cluster.encoder,
3352 push_handler_func=push_handler_func,
3353 event_dispatcher=self._event_dispatcher,
3354 **kwargs,
3355 )
3356 # Subscribe to slots-cache change notifications so shard subscriptions
3357 # can be reconciled automatically after topology refreshes.
3358 nm_dispatcher = redis_cluster.nodes_manager._event_dispatcher
3359 self._slots_cache_listener = ClusterPubSubSlotsCacheListener(self)
3360 nm_dispatcher.register_listeners(
3361 {AfterSlotsCacheRefreshEvent: [self._slots_cache_listener]}
3362 )
3363 # Deterministic GC-time cleanup so short-lived pubsubs do not leak
3364 # listeners in the dispatcher when no slots-refresh event ever fires.
3365 weakref.finalize(
3366 self,
3367 _unregister_slots_cache_listener,
3368 weakref.ref(nm_dispatcher),
3369 self._slots_cache_listener,
3370 AfterSlotsCacheRefreshEvent,
3371 )
3373 def set_pubsub_node(self, cluster, node=None, host=None, port=None):
3374 """
3375 The pubsub node will be set according to the passed node, host and port
3376 When none of the node, host, or port are specified - the node is set
3377 to None and will be determined by the keyslot of the channel in the
3378 first command to be executed.
3379 RedisClusterException will be thrown if the passed node does not exist
3380 in the cluster.
3381 If host is passed without port, or vice versa, a DataError will be
3382 thrown.
3383 :type cluster: RedisCluster
3384 :type node: ClusterNode
3385 :type host: str
3386 :type port: int
3387 """
3388 if node is not None:
3389 # node is passed by the user
3390 self._raise_on_invalid_node(cluster, node, node.host, node.port)
3391 pubsub_node = node
3392 elif host is not None and port is not None:
3393 # host and port passed by the user
3394 node = cluster.get_node(host=host, port=port)
3395 self._raise_on_invalid_node(cluster, node, host, port)
3396 pubsub_node = node
3397 elif any([host, port]) is True:
3398 # only 'host' or 'port' passed
3399 raise DataError("Passing a host requires passing a port, and vice versa")
3400 else:
3401 # nothing passed by the user. set node to None
3402 pubsub_node = None
3404 self.node = pubsub_node
3406 def get_pubsub_node(self):
3407 """
3408 Get the node that is being used as the pubsub connection
3409 """
3410 return self.node
3412 def _raise_on_invalid_node(self, redis_cluster, node, host, port):
3413 """
3414 Raise a RedisClusterException if the node is None or doesn't exist in
3415 the cluster.
3416 """
3417 if node is None or redis_cluster.get_node(node_name=node.name) is None:
3418 raise RedisClusterException(
3419 f"Node {host}:{port} doesn't exist in the cluster"
3420 )
3422 def execute_command(self, *args):
3423 """
3424 Execute a subscribe/unsubscribe command.
3426 Taken code from redis-py and tweak to make it work within a cluster.
3427 """
3428 # NOTE: don't parse the response in this function -- it could pull a
3429 # legitimate message off the stack if the connection is already
3430 # subscribed to one or more channels
3432 # For shard commands, route to appropriate node
3433 command = args[0].upper() if args else ""
3434 if command in ("SSUBSCRIBE", "SUNSUBSCRIBE", "SPUBLISH"):
3435 if len(args) > 1:
3436 # ssubscribe / sunsubscribe own both the per-node I/O lock and
3437 # the shard_channels / _shard_channel_to_node bookkeeping, so
3438 # delegate to them instead of dispatching raw. A raw dispatch
3439 # writes the socket unguarded against a concurrent poll and
3440 # records nothing, leaving the channel invisible to the reader
3441 # loop and to on_connect's replay.
3442 if command == "SSUBSCRIBE":
3443 return self.ssubscribe(*args[1:])
3444 if command == "SUNSUBSCRIBE":
3445 return self.sunsubscribe(*args[1:])
3446 channel = args[1]
3447 node = self.cluster.get_node_from_key(channel)
3448 if node:
3449 pubsub = self._get_node_pubsub(node)
3450 with self._pubsub_io_lock(pubsub):
3451 return pubsub.execute_command(*args)
3453 # For other commands, use the set node or lazily discover one
3454 if self.connection is None:
3455 if self.connection_pool is None:
3456 if len(args) > 1:
3457 # Hash the first channel and get one of the nodes holding
3458 # this slot
3459 channel = args[1]
3460 slot = self.cluster.keyslot(channel)
3461 node = self.cluster.nodes_manager.get_node_from_slot(
3462 slot,
3463 self.cluster.read_from_replicas,
3464 self.cluster.load_balancing_strategy,
3465 )
3466 else:
3467 # Get a random node
3468 node = self.cluster.get_random_node()
3469 self.node = node
3470 redis_connection = self.cluster.get_redis_connection(node)
3471 self.connection_pool = redis_connection.connection_pool
3472 self.connection = self.connection_pool.get_connection()
3473 # register a callback that re-subscribes to any channels we
3474 # were listening to when we were disconnected
3475 self.connection.register_connect_callback(self.on_connect)
3476 if self.push_handler_func is not None:
3477 self.connection._parser.set_pubsub_push_handler(self.push_handler_func)
3478 self._event_dispatcher.dispatch(
3479 AfterPubSubConnectionInstantiationEvent(
3480 self.connection, self.connection_pool, ClientType.SYNC, self._lock
3481 )
3482 )
3483 connection = self.connection
3484 self._execute(connection, connection.send_command, *args)
3486 def _resubscribe_shard_channels(self) -> None:
3487 # A single node can own multiple slot ranges, so a batched
3488 # ``SSUBSCRIBE`` covering every tracked channel would be rejected by
3489 # Redis with a ``CROSSSLOT`` error. Group by hash slot and emit one
3490 # ``SSUBSCRIBE`` per slot.
3491 by_slot: defaultdict[int, dict] = defaultdict(dict)
3492 for k, v in self.shard_channels.items():
3493 by_slot[key_slot(self.encoder.encode(k))][k] = v
3494 for subscriptions in by_slot.values():
3495 self._resubscribe(subscriptions, self.ssubscribe)
3497 def _get_node_pubsub(self, node):
3498 try:
3499 return self.node_pubsub_mapping[node.name]
3500 except KeyError:
3501 redis_connection = self.cluster.get_redis_connection(node)
3502 pubsub = redis_connection.pubsub(
3503 push_handler_func=self.push_handler_func,
3504 )
3505 # Replay shard subscriptions on reconnect with slot-aware grouping
3506 # so that channels spanning multiple slots owned by this node do
3507 # not trigger a CROSSSLOT error.
3508 pubsub._resubscribe_shard_channels = MethodType(
3509 ClusterPubSub._resubscribe_shard_channels, pubsub
3510 )
3511 self._pubsub_io_lock(pubsub)
3512 self.node_pubsub_mapping[node.name] = pubsub
3513 return pubsub
3515 def _find_node_name_for_pubsub(self, pubsub):
3516 # Snapshot the items: every caller but one runs without
3517 # _shard_state_lock, so iterating the mapping directly would raise
3518 # "dictionary changed size during iteration" whenever a concurrent
3519 # migration adds or retires a per-node pubsub - which is exactly when
3520 # these lookups happen. list() of a dict view is atomic under the GIL.
3521 for node_name, node_pubsub in list(self.node_pubsub_mapping.items()):
3522 if node_pubsub is pubsub:
3523 return node_name
3524 return None
3526 @staticmethod
3527 def _pubsub_io_lock(pubsub) -> threading.RLock:
3528 """Return the per-node pubsub's wire I/O lock, creating it on first use.
3530 A per-node ``PubSub`` is read by whichever thread polls
3531 ``get_sharded_message`` and written by the reconciliation worker
3532 (``_migrate_shard_channel``) and by any caller of ``ssubscribe`` /
3533 ``sunsubscribe``. ``PubSub`` guards writes with its own ``_lock``
3534 (``PubSub.execute_command``) but reads take no lock at all, so without
3535 this the reader can be inside ``read_response`` while another thread's
3536 ``_execute`` disconnects and reconnects the same socket underneath it -
3537 which loses the reply to the handshake and surfaces as a read timeout
3538 followed by ``EBADF``.
3540 Kept on the pubsub rather than in a dict keyed by node name so it
3541 travels with the object through ``node_pubsub_mapping`` and cannot go
3542 stale when a per-node pubsub is dropped and recreated.
3543 """
3544 lock = getattr(pubsub, "_shard_io_lock", None)
3545 if lock is None:
3546 lock = threading.RLock()
3547 pubsub._shard_io_lock = lock
3548 return lock
3550 @staticmethod
3551 def _detach_shard_channel(pubsub, channel) -> None:
3552 """Forget a shard channel on a per-node pubsub without a wire round trip.
3554 ``PubSub.sunsubscribe`` only records the intent in
3555 ``pending_unsubscribe_shard_channels``; the channel leaves
3556 ``shard_channels`` when the server confirmation is read. So if the
3557 ``SUNSUBSCRIBE`` never reaches the server, ``on_connect`` clears the
3558 pending set and replays ``SSUBSCRIBE`` for the channel - on the node it
3559 is being migrated away from, on every reconnect. Once the caller has
3560 decided the channel belongs to a different node, the local intent is
3561 the only truth left, so drop it here.
3563 Deliberately without the per-node I/O lock, unlike every other writer
3564 on a per-node pubsub. ``_forget_shard_channel_on_old_node`` calls this
3565 for a node the reader has just failed to reach, and a bounded poll holds
3566 that lock across ``PubSub._execute``'s reconnect and its whole retry
3567 budget - far longer than the timeout the poll was given. Waiting for it
3568 here, while this pass holds ``_shard_state_lock``, is the migration
3569 slow enough to look like a permanent delivery stall that the
3570 ``_unreachable_nodes`` fast path exists to avoid.
3572 Nothing needs the lock: each mutation below is individually atomic, and
3573 every interleaving with ``handle_message``'s unsubscribe bookkeeping (the
3574 only concurrent writer of the same state, and one that runs under that
3575 lock) converges on the same end state - the channel gone from both the
3576 subscription dict and the pending set. That bookkeeping discards rather
3577 than removes precisely so this detach cannot make it raise ``KeyError``
3578 into a poll no caller catches.
3579 """
3580 pubsub.shard_channels.pop(channel, None)
3581 pubsub.pending_unsubscribe_shard_channels.discard(channel)
3582 if not pubsub.channels and not pubsub.patterns and not pubsub.shard_channels:
3583 pubsub.subscribed_event.clear()
3585 def _drop_node_pubsub(self, name: str, pubsub) -> None:
3586 """Retire a per-node pubsub and drop it from ``node_pubsub_mapping``.
3588 Callers hold ``_shard_state_lock``, the lock that every mutation of
3589 that mapping observes. ``reset()`` runs under the per-node I/O lock so
3590 the socket is not torn down beneath a concurrent bounded poll parked in
3591 ``parse_response``, and its errors are swallowed: retiring one node's
3592 pubsub must not abort the caller's pass, and this also runs from the
3593 ``__del__`` fallback path through ``reset()``.
3595 Every caller must leave nothing subscribed on ``pubsub`` (or have lost
3596 the node itself). An empty per-node pubsub left in the mapping is what
3597 ``_poll_node_pubsub`` stalls on: its ``subscribed_event`` is cleared, so
3598 the prelude waits out the whole timeout of every pass - indefinitely when
3599 the caller passed ``timeout=None``, since the ``reset()`` below only
3600 clears that event again and no message can arrive on a subscription that
3601 is gone.
3603 Popping it from the mapping is also what makes the retirement
3604 *observable*: that prelude gives up on an unbounded wait once the pubsub
3605 it was handed is no longer mapped, which is the only thing standing
3606 between a reader that held the object across this call and a permanent
3607 park. The rebind below keeps it from being handed out to begin with.
3608 """
3609 try:
3610 with self._pubsub_io_lock(pubsub):
3611 pubsub.reset()
3612 except Exception:
3613 pass
3614 self.node_pubsub_mapping.pop(name, None)
3615 self._unreachable_nodes.discard(name)
3616 # Same snapshot reason ``reset()`` recreates this: ``_pubsubs_generator``
3617 # captures node_pubsub_mapping.values() into a local list inside
3618 # ``yield from``, which the pop above does not reach - so a generator
3619 # suspended mid-yield-from would still hand the object we just retired
3620 # to the next poll. ``type(self)`` bypasses the instance-level
3621 # self-shadow established at __init__. Costs nothing: constructing a
3622 # generator runs no frame, so the per-node collection loop in
3623 # reinitialize_shard_subscriptions can rebind once per dropped node.
3624 self._pubsubs_generator = type(self)._pubsubs_generator(self)
3626 def _sharded_message_generator(self, timeout=0.0):
3627 first_error: Optional[BaseException] = None
3628 polled = 0
3629 failed = 0
3630 next_ready: Optional[float] = None
3631 for _ in range(len(self.node_pubsub_mapping)):
3632 pubsub = next(self._pubsubs_generator)
3633 if pubsub is None:
3634 # node_pubsub_mapping was emptied between the len() above and
3635 # here; nothing left to poll in this pass.
3636 break
3637 if self._poll_cool_off:
3638 deadline = self._poll_cool_off.get(pubsub, 0.0)
3639 if time.monotonic() < deadline:
3640 # In cool-off after a failed poll: skip it so the reader
3641 # spends this pass on the nodes that can still deliver.
3642 if next_ready is None or deadline < next_ready:
3643 next_ready = deadline
3644 continue
3645 polled += 1
3646 try:
3647 message = self._poll_node_pubsub(pubsub, timeout)
3648 except MovedError as e:
3649 # Handled, not failed: _handle_moved_on_read re-routes the
3650 # offending channels and schedules reconciliation, so the next
3651 # pass recovers. Re-raising a MovedError out of a pubsub read
3652 # would only hand the caller an error it cannot act on. Still
3653 # cool off: if the slots cache cannot be corrected the repair
3654 # would otherwise re-run on every poll.
3655 self._poll_cool_off[pubsub] = (
3656 time.monotonic() + SHARD_POLL_COOL_OFF_SECONDS
3657 )
3658 self._handle_moved_on_read(pubsub, e)
3659 continue
3660 except (ConnectionError, TimeoutError, OSError) as e:
3661 # One unhealthy node must not starve its healthy siblings. A
3662 # single reader serves every per-node pubsub, so aborting the
3663 # pass here stops delivery cluster-wide for as long as this one
3664 # node stays unreachable - even though the slots it no longer
3665 # serves are the only ones affected. Keep polling the rest and
3666 # surface an error only if nothing in the pass worked, the same
3667 # made-progress rule reinitialize_shard_subscriptions applies.
3668 failed += 1
3669 if first_error is None:
3670 first_error = e
3671 # Cool off before polling this one again. PubSub._execute
3672 # reconnects and then retries through the connection's own
3673 # Retry, so a single "bounded" poll on an unreachable node can
3674 # cost its whole retry budget - far longer than the timeout the
3675 # caller asked for. Without a cool-off the reader goes straight
3676 # back to that node on the next pass and pays it again, which
3677 # is what turns one sick node into a cluster-wide delivery
3678 # stall.
3679 self._poll_cool_off[pubsub] = (
3680 time.monotonic() + SHARD_POLL_COOL_OFF_SECONDS
3681 )
3682 # Deliberately outside _shard_state_lock: this is an advisory
3683 # hint for _migrate_shard_channel's fast path, mutated with
3684 # single atomic set operations, and both misread directions are
3685 # handled there and self-heal - a stale entry only skips a
3686 # SUNSUBSCRIBE to a dead node, a missing one only pays a
3687 # reconnect before the same local forget.
3688 node_name = self._find_node_name_for_pubsub(pubsub)
3689 if node_name is not None:
3690 self._unreachable_nodes.add(node_name)
3691 if is_debug_log_enabled():
3692 logger.debug(
3693 "sharded pubsub poll failed on %s: %s: %s",
3694 node_name,
3695 type(e).__name__,
3696 e,
3697 )
3698 # A node that has left the deployment never answers MOVED, so
3699 # this branch is the only signal that its shard channels may
3700 # need a new owner. Ask for a slots-cache refresh; its dispatch
3701 # reaches on_slots_changed and reconciles.
3702 self._schedule_topology_repair()
3703 continue
3704 # Emptiness check first: this is the per-message hot path, and the
3705 # weakref lookup a WeakKeyDictionary pop needs is pure overhead
3706 # while no node is in cool-off, which is the normal case.
3707 if self._poll_cool_off:
3708 self._poll_cool_off.pop(pubsub, None)
3709 if self._unreachable_nodes:
3710 node_name = self._find_node_name_for_pubsub(pubsub)
3711 if node_name is not None:
3712 self._unreachable_nodes.discard(node_name)
3713 if message is not None:
3714 return pubsub, message
3715 if first_error is not None and failed == polled:
3716 raise first_error
3717 if polled == 0 and next_ready is not None:
3718 self._wait_out_cool_off(next_ready, timeout)
3719 return None, None
3721 @staticmethod
3722 def _wait_out_cool_off(next_ready: float, timeout) -> None:
3723 """Wait out a pass in which every node was skipped for cool-off.
3725 Such a pass does no wire read at all, so returning straight away
3726 ignores the timeout the caller asked to block for - and a reader loop
3727 (``PubSubWorkerThread.run`` with ``sharded_pubsub=True``, or a user
3728 loop on ``get_sharded_message``) polls back immediately, spinning
3729 until the cool-off expires instead of blocking. Sleep instead: until
3730 the earliest cool-off is over, never longer than the caller's timeout,
3731 and not at all for a non-blocking poll.
3732 """
3733 if timeout is not None and timeout <= 0:
3734 return
3735 delay = next_ready - time.monotonic()
3736 if delay <= 0:
3737 return
3738 if timeout is not None:
3739 delay = min(delay, timeout)
3740 time.sleep(delay)
3742 def _poll_io_lock(self, pubsub, timeout):
3743 """Guard a per-node poll against concurrent writers on the same socket.
3745 ``timeout=None`` makes ``_poll_node_pubsub``'s read block indefinitely,
3746 so holding the lock across it would block reconciliation for as long as
3747 no message arrives. Such a caller drives the pubsub itself and gets the
3748 pre-existing unguarded behavior; every bounded poll - which is what
3749 ``PubSubWorkerThread`` and ``ClusterPubSub``'s own callers use - is
3750 serialized.
3751 """
3752 if timeout is None:
3753 return nullcontext()
3754 return self._pubsub_io_lock(pubsub)
3756 def _poll_node_pubsub(self, pubsub, timeout):
3757 """Read one message from a per-node pubsub, dispatching outside the lock.
3759 Splits ``PubSub.get_message`` so the per-node I/O lock covers the wire
3760 read only. ``handle_message`` invokes a subscribed channel's user
3761 handler inline, and a handler is free to call ``ssubscribe`` /
3762 ``sunsubscribe`` on this ``ClusterPubSub`` - which takes
3763 ``_shard_state_lock`` and then the same I/O lock. Holding the I/O lock
3764 across the handler therefore deadlocks against the reconciliation
3765 worker, which holds ``_shard_state_lock`` and waits for that I/O lock:
3766 an ABBA cycle between two threads that the ``RLock``'s self-reentrancy
3767 cannot break. The async counterpart's non-reentrant ``asyncio.Lock``
3768 hangs on the re-acquire alone, before any reconciliation is involved.
3770 The two halves of ``handle_message`` are mutually exclusive:
3771 ``UNSUBSCRIBE_MESSAGE_TYPES`` does subscription bookkeeping and never
3772 reaches a handler, ``PUBLISH_MESSAGE_TYPES`` only dispatches. So
3773 bookkeeping stays inside the lock - it mutates the very
3774 ``shard_channels`` / ``pending_unsubscribe_shard_channels`` that
3775 ``ssubscribe`` / ``sunsubscribe`` mutate under this lock - and only the
3776 dispatch moves out. The cost is a narrow race: a reconciliation pass
3777 that detaches the channel between the read and the dispatch makes the
3778 handler lookup miss, so the message is returned to the caller instead
3779 of dispatched. That is the same in-flight-during-unsubscribe race
3780 ``PubSub`` itself has, and closing it would mean duplicating
3781 ``handle_message``'s dispatch here.
3782 """
3783 # ``PubSub.get_message``'s prelude, which bypassing it would drop: a
3784 # per-node pubsub enters node_pubsub_mapping before its first
3785 # SSUBSCRIBE (_get_node_pubsub) and is left connectionless by reset()
3786 # (the GC in reinitialize_shard_subscriptions), while parse_response
3787 # raises RuntimeError on a None connection - which neither poll site
3788 # catches. Deliberately kept outside the I/O lock, unlike the
3789 # get_message call it replaces: waiting here is not wire I/O, and the
3790 # ssubscribe that sets this event needs the I/O lock itself, so waiting
3791 # under it stalls the very subscribe being waited for.
3792 if not pubsub.subscribed:
3793 start_time = time.monotonic()
3794 # An unbounded caller must not wait on this event uninterrupted.
3795 # _drop_node_pubsub retires a per-node pubsub whose event is cleared
3796 # and will never be set again - it is never resubscribed - and the
3797 # reader can be holding that object across the drop: the round robin
3798 # yields from a _pubsubs_generator snapshot the drop's pop does not
3799 # reach, and get_sharded_message's target_node lookup takes no lock.
3800 # So wake up periodically and re-check the mapping rather than park
3801 # the single reader for good, which would withhold the messages of
3802 # every healthy sibling in the pass too.
3803 if timeout is None:
3804 wait_for = SHARD_SUBSCRIBE_WAIT_TICK_SECONDS
3805 else:
3806 wait_for = timeout
3807 while not pubsub.subscribed_event.wait(wait_for):
3808 if timeout is not None:
3809 # The connection isn't subscribed to any channels or
3810 # patterns, so no messages are available
3811 return None
3812 if self._find_node_name_for_pubsub(pubsub) is None:
3813 # Retired by a concurrent drop while we held it. Nothing can
3814 # arrive on a subscription that is gone; let the pass move on
3815 # to the nodes that can still deliver.
3816 return None
3817 # The connection was subscribed during the timeout time frame.
3818 # The timeout should be adjusted based on the time spent
3819 # waiting for the subscription. Only the bounded case needs it: an
3820 # unbounded wait leaves ``timeout`` at None for the read below.
3821 if timeout is not None:
3822 timeout = max(0.0, timeout - (time.monotonic() - start_time))
3823 with self._poll_io_lock(pubsub, timeout):
3824 # Re-check now that no writer can be mid-flight: the prelude above
3825 # can pass and the GC then close this pubsub while holding this very
3826 # lock, which would leave parse_response with the None connection it
3827 # raises RuntimeError on. Only the connection is re-checked, not
3828 # ``subscribed``: a pubsub whose last channel was detached locally
3829 # may still have an in-flight SUNSUBSCRIBE confirmation to read.
3830 if pubsub.connection is None:
3831 return None
3832 response = pubsub.parse_response(block=(timeout is None), timeout=timeout)
3833 # get_message's truthiness test, not "is None": a health check
3834 # reply filtered out by parse_response, or an empty bulk, is "no
3835 # message" rather than a message to parse.
3836 if not response:
3837 return None
3838 if not self._is_publish_response(response):
3839 # Don't pass ignore_subscribe_messages here - let
3840 # get_sharded_message handle the filtering after processing
3841 # subscription state changes
3842 return pubsub.handle_message(response, ignore_subscribe_messages=False)
3843 return pubsub.handle_message(response, ignore_subscribe_messages=False)
3845 @staticmethod
3846 def _is_publish_response(response) -> bool:
3847 """Whether a raw pubsub reply can make ``handle_message`` dispatch.
3849 ``handle_message`` invokes a user handler only for
3850 ``PUBLISH_MESSAGE_TYPES``; every other reply either does subscription
3851 bookkeeping (``UNSUBSCRIBE_MESSAGE_TYPES``) or is a pong, and the two
3852 branches are mutually exclusive. A non-sequence reply is the bare-PING
3853 shape ``handle_message`` rewrites into a pong, so it cannot dispatch
3854 either.
3855 """
3856 if not isinstance(response, (list, tuple)):
3857 return False
3858 return str_if_bytes(response[0]) in PubSub.PUBLISH_MESSAGE_TYPES
3860 def _schedule_topology_repair(self) -> None:
3861 """Ask for a slots-cache refresh after a poll could not reach a node.
3863 ``reinitialize_shard_subscriptions`` only ever runs from a slots-cache
3864 change notification, and a node that has been rebooted or taken out of
3865 the deployment answers ``ECONNREFUSED`` rather than ``MOVED`` - so the
3866 read path itself has to ask, or the shard channels pinned to that node
3867 stay there for the lifetime of the pubsub.
3869 ``NodesManager.initialize`` serializes concurrent callers, drops nodes
3870 that have left the topology and dispatches
3871 ``AfterSlotsCacheRefreshEvent``, which reaches ``on_slots_changed``; run
3872 it on the reconciliation worker so a bounded poll does not pay for a
3873 ``CLUSTER SLOTS`` round trip, and throttle it because a node that is
3874 down fails every poll.
3876 ``disconnect_startup_nodes_pools=False``, unlike every other caller: the
3877 default marks the in-use connections of the startup node that answered
3878 ``CLUSTER SLOTS`` for reconnect and drops its idle ones, to clear a
3879 ``READONLY`` state that may no longer apply. A per-node pubsub draws its
3880 connection from that very pool, so on the default one unreachable shard
3881 would recycle command and pubsub connections on a *healthy* node once
3882 per throttle window - forcing a resubscribe and dropping in-flight
3883 messages on a node that never failed. A fresh slot map is all this
3884 repair asks for. The async counterpart has no such parameter and never
3885 recycles those pools, so this is also the shape parity asks for.
3886 """
3887 if not self.shard_channels:
3888 return
3889 now = time.monotonic()
3890 if now < self._next_topology_repair:
3891 return
3892 # Unsynchronized on purpose: the read path must not queue behind a
3893 # reconciliation pass doing per-channel network I/O, because the
3894 # cluster-wide delivery stall that would cause is the very thing this
3895 # repair exists to end. Two readers racing the throttle window can both
3896 # submit; the single-worker executor runs them sequentially, so the
3897 # loser costs one redundant CLUSTER SLOTS round trip at most (the epoch
3898 # check in NodesManager.initialize dedups only overlapping refreshes,
3899 # not queued ones), bounded by the throttle window.
3900 self._next_topology_repair = now + SHARD_TOPOLOGY_REPAIR_INTERVAL_SECONDS
3901 self._submit_reconcile_work(
3902 partial(
3903 self.cluster.nodes_manager.initialize,
3904 disconnect_startup_nodes_pools=False,
3905 )
3906 )
3908 def _handle_moved_on_read(self, pubsub, error: MovedError) -> None:
3909 """Re-route shard channels pinned to a node that lost their slot.
3911 ``PubSub.on_connect`` replays ``SSUBSCRIBE`` to the node its connection
3912 is bound to, so after a slot migration that node answers ``MOVED``.
3913 ``MovedError`` is not in ``Retry.supported_errors`` and no other code on
3914 the read path refreshes the slots cache, so a shard channel left on a
3915 former owner could never recover. Drop the offending channels from this
3916 pubsub so the replay stops, forget their recorded owner so
3917 ``reinitialize_shard_subscriptions`` does not short-circuit on an
3918 already-advanced reverse index, then apply the redirect and reconcile.
3919 """
3920 node_name = self._find_node_name_for_pubsub(pubsub)
3921 logger.debug(
3922 "sharded pubsub: %s no longer owns slot %s; re-routing its shard channels",
3923 node_name,
3924 error.slot_id,
3925 )
3926 with self._shard_state_lock:
3927 for channel in list(pubsub.shard_channels):
3928 if key_slot(self.encoder.encode(channel)) != error.slot_id:
3929 continue
3930 self._detach_shard_channel(pubsub, channel)
3931 if self._shard_channel_to_node.get(channel) == node_name:
3932 del self._shard_channel_to_node[channel]
3933 # The detach above can leave this pubsub with nothing subscribed -
3934 # a node that lost its only slot answers MOVED for every channel it
3935 # held. Retire it here rather than leave it in the mapping for a
3936 # collector elsewhere: no SUNSUBSCRIBE confirmation will arrive for
3937 # a channel forgotten locally, so get_sharded_message's collector
3938 # cannot reach it, and the reconciliation pass scheduled below only
3939 # GCs it once the worker gets to run - a whole poll cool-off later,
3940 # at best, while an unbounded poll that reaches the empty pubsub
3941 # first parks on it for good (see _drop_node_pubsub).
3942 if node_name is not None and not pubsub.subscribed:
3943 self._drop_node_pubsub(node_name, pubsub)
3944 # move_slot applies the redirect to the slots cache and dispatches
3945 # AfterSlotsCacheRefreshEvent, which reaches on_slots_changed. Call
3946 # on_slots_changed unconditionally too: move_slot skips the dispatch on
3947 # a circular MOVED, and a duplicate reconciliation pass is a no-op.
3948 # move_slot indexes slots_cache by the redirected slot, so an
3949 # as-yet-uncovered slot raises: log and still reconcile rather than let
3950 # a repair attempt break a pubsub read.
3951 try:
3952 self.cluster.nodes_manager.move_slot(error)
3953 except Exception as exc:
3954 logger.debug(
3955 "sharded pubsub: could not apply the redirect for slot %s: %s: %s",
3956 error.slot_id,
3957 type(exc).__name__,
3958 exc,
3959 )
3960 self.on_slots_changed()
3962 def _pubsubs_generator(self):
3963 # Never return: a generator that returns is exhausted for good and only
3964 # reset() recreates this one, so a momentarily empty
3965 # node_pubsub_mapping - reconciliation drops a per-node pubsub before
3966 # creating its replacement - would stop the round robin permanently.
3967 # Yield None for an empty mapping instead, which lets the caller skip
3968 # the slot without this loop spinning on an empty list.
3969 while True:
3970 current_nodes = list(self.node_pubsub_mapping.values())
3971 if not current_nodes:
3972 yield None
3973 else:
3974 yield from current_nodes
3976 def get_sharded_message(
3977 self, ignore_subscribe_messages=False, timeout=0.0, target_node=None
3978 ):
3979 """
3980 Get the next sharded pubsub message, or ``None`` if none is available.
3982 Polls the per-node connections in round robin unless ``target_node`` is
3983 given, and keeps shard channels attached to the node that currently
3984 owns their slot: a failed poll cools that node off and asks for a
3985 slots-cache refresh, and a ``MOVED`` reply re-routes the affected
3986 channels to their new owner. Neither reaches the caller. A connection
3987 failure is surfaced only when every node polled in the pass failed, so
3988 one unreachable node does not stop delivery from its healthy siblings.
3990 ``target_node`` opts out of that shielding: a caller that names a
3991 single node has no sibling to protect, so connection errors propagate.
3992 A ``MOVED`` reply is still handled rather than raised.
3994 :param ignore_subscribe_messages: Whether to ignore subscribe messages
3995 :param timeout: Timeout for message retrieval
3996 :param target_node: Specific node to get message from
3997 :return: Message dictionary or None
3998 """
3999 if target_node:
4000 # Use .get(): migration-driven cleanup in the sunsubscribe branch
4001 # below and reset() both remove entries from node_pubsub_mapping,
4002 # so a caller polling with target_node may race the cleanup. Match
4003 # the async counterpart's None-handling rather than raising
4004 # KeyError. None pubsub falls through to "no message available".
4005 pubsub = self.node_pubsub_mapping.get(target_node.name)
4006 if pubsub is not None:
4007 try:
4008 message = self._poll_node_pubsub(pubsub, timeout)
4009 except MovedError as e:
4010 # Same handling as the round-robin path: the caller cannot
4011 # act on a MovedError raised out of a pubsub read, and the
4012 # channels this node no longer owns have to be re-routed or
4013 # they never recover. Cool off too, so a slots cache that
4014 # cannot be corrected does not re-run the repair on every
4015 # poll. Unlike that path, connectivity errors still
4016 # propagate: they are swallowed there only to keep one sick
4017 # node from starving its healthy siblings, and a caller that
4018 # named a single node has no sibling to protect.
4019 self._poll_cool_off[pubsub] = (
4020 time.monotonic() + SHARD_POLL_COOL_OFF_SECONDS
4021 )
4022 self._handle_moved_on_read(pubsub, e)
4023 message = None
4024 else:
4025 message = None
4026 else:
4027 pubsub, message = self._sharded_message_generator(timeout=timeout)
4028 if message is None:
4029 return None
4030 # Only sunsubscribe mutates cluster-level shard state; bypassing the
4031 # lock on the data-message hot path keeps smessage delivery from
4032 # competing with the reconciliation worker for _shard_state_lock.
4033 if str_if_bytes(message["type"]) == "sunsubscribe":
4034 # Serialize state mutation against reinitialize_shard_subscriptions
4035 # (worker thread). The blocking _poll_node_pubsub above
4036 # intentionally runs outside the lock so reconciliation is not
4037 # stalled by long polls.
4038 with self._shard_state_lock:
4039 if message["channel"] in self.pending_unsubscribe_shard_channels:
4040 # User-initiated sunsubscribe: drop from cluster-level tracking.
4041 self.pending_unsubscribe_shard_channels.remove(message["channel"])
4042 self.shard_channels.pop(message["channel"], None)
4043 self._shard_channel_to_node.pop(message["channel"], None)
4044 # Drop the per-node pubsub that delivered the confirmation once
4045 # it no longer holds any shard subscriptions, regardless of
4046 # whether the sunsubscribe was user-initiated or driven by
4047 # slot-migration reconciliation (_migrate_shard_channel, which
4048 # intentionally does not add the channel to
4049 # pending_unsubscribe_shard_channels). This releases the
4050 # dedicated connection that would otherwise linger.
4051 # Identifying the receiving pubsub directly (rather than via
4052 # the cluster's current slot map) is required after slot
4053 # migration, where the channel's owner is no longer the node
4054 # that received our original SSUBSCRIBE.
4055 if pubsub is not None and not pubsub.subscribed:
4056 name = self._find_node_name_for_pubsub(pubsub)
4057 if name is not None:
4058 self._drop_node_pubsub(name, pubsub)
4059 # Mirror PubSub.handle_message: the empty-check belongs in the
4060 # unsubscribe branch since that is the only path that can
4061 # reduce shard_channels here.
4062 if not self.channels and not self.patterns and not self.shard_channels:
4063 self.subscribed_event.clear()
4064 # Only suppress subscribe/unsubscribe messages, not data messages (smessage)
4065 if str_if_bytes(message["type"]) in ("ssubscribe", "sunsubscribe"):
4066 if self.ignore_subscribe_messages or ignore_subscribe_messages:
4067 return None
4068 return message
4070 def ssubscribe(
4071 self, *args: ChannelT | Subscription, **kwargs: PubSubHandler
4072 ) -> None:
4073 """
4074 Subscribe to shard channels.
4076 Channels supplied as keyword arguments expect a channel name as the key
4077 and a callable as the value. ``Subscription`` objects can also be
4078 supplied positionally with an optional handler.
4079 """
4080 s_channels = parse_pubsub_subscriptions(args, kwargs)
4081 # Serialize against reinitialize_shard_subscriptions (worker thread)
4082 # so the reverse index, shard_channels, and node_pubsub_mapping are
4083 # not mutated concurrently.
4084 with self._shard_state_lock:
4085 for s_channel, handler in s_channels.items():
4086 node = self.cluster.get_node_from_key(s_channel)
4087 if not node:
4088 continue
4089 # Lazy re-route: if this channel is already tracked against a
4090 # different node (e.g. after a slot migration), migrate it now
4091 # so the caller's intent is applied on the current owner.
4092 normalized_key = next(iter(self._normalize_keys({s_channel: None})))
4093 old_name = self._shard_channel_to_node.get(normalized_key)
4094 if old_name and old_name != node.name:
4095 # Match PubSub.ssubscribe() dict.update() semantics: the
4096 # caller's newly supplied handler (including None) always
4097 # overrides any previously registered handler.
4098 self._migrate_shard_channel(
4099 normalized_key,
4100 handler,
4101 old_name,
4102 node,
4103 )
4104 continue
4105 pubsub = self._get_node_pubsub(node)
4106 with self._pubsub_io_lock(pubsub):
4107 if handler:
4108 pubsub.ssubscribe(Subscription(s_channel, handler))
4109 else:
4110 pubsub.ssubscribe(s_channel)
4111 self.shard_channels.update(pubsub.shard_channels)
4112 self._shard_channel_to_node[normalized_key] = node.name
4113 self.pending_unsubscribe_shard_channels.difference_update(
4114 self._normalize_keys({s_channel: None})
4115 )
4116 if pubsub.subscribed and not self.subscribed:
4117 self.subscribed_event.set()
4118 self.health_check_response_counter = 0
4120 def sunsubscribe(self, *args):
4121 if args:
4122 args = list_or_args(args[0], args[1:])
4123 else:
4124 args = list(self.shard_channels)
4126 # Serialize against reinitialize_shard_subscriptions: the reverse
4127 # index and node_pubsub_mapping must not change between the lookup
4128 # and the per-node sunsubscribe call below.
4129 with self._shard_state_lock:
4130 for s_channel in args:
4131 normalized_key = next(iter(self._normalize_keys({s_channel: None})))
4132 # Route via the reverse index so we unsubscribe on the node
4133 # that actually holds the subscription. After a slot migration
4134 # the cluster's current owner may no longer be that node.
4135 name = self._shard_channel_to_node.get(normalized_key)
4136 if name and name in self.node_pubsub_mapping:
4137 p = self.node_pubsub_mapping[name]
4138 else:
4139 node = self.cluster.get_node_from_key(s_channel)
4140 if not node or node.name not in self.node_pubsub_mapping:
4141 continue
4142 p = self.node_pubsub_mapping[node.name]
4143 with self._pubsub_io_lock(p):
4144 p.sunsubscribe(s_channel)
4145 self.pending_unsubscribe_shard_channels.update(
4146 p.pending_unsubscribe_shard_channels
4147 )
4149 def reinitialize_shard_subscriptions(self):
4150 """
4151 Reconcile per-node shard subscriptions against the cluster's current
4152 slot ownership map. For each tracked shard channel whose owning node
4153 has changed (e.g. after CLUSTER SETSLOT / failover), sunsubscribe on
4154 the old node's pubsub and ssubscribe on the new owner's pubsub,
4155 preserving any registered handler.
4156 """
4157 uncovered: list = []
4158 made_progress = False
4159 first_migrate_error: Optional[BaseException] = None
4160 with self._shard_state_lock:
4161 for channel, handler in list(self.shard_channels.items()):
4162 if channel in self.pending_unsubscribe_shard_channels:
4163 continue
4164 try:
4165 new_node = self.cluster.get_node_from_key(channel)
4166 except SlotNotCoveredError:
4167 # Slot is transiently uncovered (mid-migration / partial
4168 # topology refresh). Defer this channel so coverable
4169 # siblings still reconcile this pass; we surface the
4170 # error below so the caller (and logs) know not every
4171 # channel was reconciled. Retry happens on the next
4172 # slots-cache change notification.
4173 uncovered.append(channel)
4174 continue
4175 old_name = self._shard_channel_to_node.get(channel)
4176 if old_name == new_node.name:
4177 owner = self.node_pubsub_mapping.get(new_node.name)
4178 if owner is not None and channel in owner.shard_channels:
4179 continue
4180 # The reverse index names this node but the subscription is
4181 # not there. _migrate_shard_channel detaches from the old
4182 # owner before it advances the index, so a pass that failed
4183 # to attach leaves the channel subscribed nowhere - and once
4184 # ownership moves back, this short-circuit would skip it for
4185 # the lifetime of the pubsub. Re-attach instead of trusting
4186 # the index; there is nothing to sunsubscribe from.
4187 old_name = None
4188 try:
4189 self._migrate_shard_channel(channel, handler, old_name, new_node)
4190 made_progress = True
4191 except (ConnectionError, TimeoutError, OSError) as e:
4192 # Transient connectivity error while subscribing on the
4193 # new owner (or unsubscribing on the old owner if its
4194 # handler chose to re-raise). Do not abort reconciliation
4195 # for sibling channels: _shard_channel_to_node was not
4196 # advanced for this channel, so the next slots-cache
4197 # change notification will retry it.
4198 logger.warning(
4199 "shard channel %r migration deferred: %s: %s",
4200 channel,
4201 type(e).__name__,
4202 e,
4203 )
4204 if first_migrate_error is None:
4205 first_migrate_error = e
4206 continue
4207 # Garbage-collect per-node pubsubs that no longer hold any
4208 # subscription so their connections are released.
4209 for name, pubsub in list(self.node_pubsub_mapping.items()):
4210 if not pubsub.subscribed:
4211 self._drop_node_pubsub(name, pubsub)
4212 if uncovered:
4213 # Surface the uncovered channels so the caller (and observer
4214 # notification path) knows reconciliation was incomplete. All
4215 # coverable siblings have already been migrated above.
4216 raise SlotNotCoveredError(
4217 f"{len(uncovered)} shard channel(s) left unreconciled; "
4218 f"slot(s) not covered by the cluster: {uncovered!r}"
4219 )
4220 if first_migrate_error is not None and not made_progress:
4221 # Every migration attempted in this pass failed transiently and
4222 # nothing else made progress. Re-raise the first caught error
4223 # (typically the root cause; later failures are often downstream
4224 # symptoms of the same unreachable node) so the worker's done-
4225 # callback surfaces a single representative failure through the
4226 # same logger channel used for SlotNotCoveredError. Per-channel
4227 # WARNINGs above preserve the full forensic detail.
4228 raise first_migrate_error
4230 def _forget_shard_channel_on_old_node(self, old_pubsub, channel, old_name):
4231 """Drop a migrating shard channel from a node we could not tell about it.
4233 Forget the channel locally: the caller advances the reverse index to the
4234 new owner, so reconciliation will never revisit this channel, while
4235 ``on_connect`` would keep replaying ``SSUBSCRIBE`` for it to this very
4236 node on every reconnect - the server would answer ``MOVED`` and the
4237 subscription would never work again.
4238 """
4239 self._detach_shard_channel(old_pubsub, channel)
4240 # Drop the per-node pubsub when either the old node has left the cluster
4241 # topology - no reconnect target, so the round-robin generator must stop
4242 # yielding a dead one, and any sibling subscription it still holds
4243 # recovers through ``PubSub._execute``'s reconnect and ``on_connect``
4244 # replay - or the detach above left it with nothing subscribed.
4245 #
4246 # The empty case cannot be deferred to a collector elsewhere, because
4247 # neither of the other two can reach it. ``get_sharded_message``'s
4248 # unsubscribe branch needs a ``SUNSUBSCRIBE`` confirmation, and none will
4249 # arrive for a channel this method forgot locally - that is the whole
4250 # reason it is forgotten. ``reinitialize_shard_subscriptions``'s
4251 # end-of-pass GC only runs for the reconciliation caller, while
4252 # ``ssubscribe``'s lazy re-route reaches here without it. An empty pubsub
4253 # left in the mapping has had its ``subscribed_event`` cleared by the
4254 # detach, so ``_poll_node_pubsub`` waits on an event nothing will ever
4255 # set: forever when the caller passed ``timeout=None``, and for the whole
4256 # timeout of every pass otherwise, before a single healthy node is read.
4257 if (
4258 self.cluster.get_node(node_name=old_name) is None
4259 or not old_pubsub.subscribed
4260 ):
4261 self._drop_node_pubsub(old_name, old_pubsub)
4263 def _migrate_shard_channel(self, channel, handler, old_name, new_node):
4264 # Detach from the old per-node pubsub, best-effort: the old node may
4265 # already be unreachable during migration / failover.
4266 if old_name and old_name in self.node_pubsub_mapping:
4267 old_pubsub = self.node_pubsub_mapping[old_name]
4268 if old_name in self._unreachable_nodes:
4269 # The reader has just failed to reach this node, so a
4270 # ``SUNSUBSCRIBE`` cannot arrive. Skip it: the attempt would pay
4271 # a full reconnect (and the client's whole retry budget) behind
4272 # the reader on the same per-node io lock, all while this pass
4273 # holds ``_shard_state_lock`` - which is what turns one departed
4274 # node into a migration slow enough to look like a permanent
4275 # delivery stall.
4276 self._forget_shard_channel_on_old_node(old_pubsub, channel, old_name)
4277 else:
4278 try:
4279 with self._pubsub_io_lock(old_pubsub):
4280 old_pubsub.sunsubscribe(channel)
4281 except (ConnectionError, TimeoutError, OSError):
4282 # redis-py's Connection has already called ``disconnect()``
4283 # before raising (see Connection.read_response /
4284 # send_packed_command with ``disconnect_on_error=True``), so
4285 # ``old_pubsub``'s dedicated socket is gone and the
4286 # ``SUNSUBSCRIBE`` never reached the server.
4287 self._forget_shard_channel_on_old_node(
4288 old_pubsub, channel, old_name
4289 )
4290 # Attach to the new per-node pubsub, preserving the handler. Decode to
4291 # a text key only when we must pass it as a kwarg (handler present).
4292 new_pubsub = self._get_node_pubsub(new_node)
4293 with self._pubsub_io_lock(new_pubsub):
4294 if handler:
4295 new_pubsub.ssubscribe(Subscription(channel, handler))
4296 else:
4297 new_pubsub.ssubscribe(channel)
4298 self.shard_channels.update(new_pubsub.shard_channels)
4299 normalized_key = next(iter(self._normalize_keys({channel: None})))
4300 self._shard_channel_to_node[normalized_key] = new_node.name
4301 self.pending_unsubscribe_shard_channels.difference_update(
4302 self._normalize_keys({channel: None})
4303 )
4304 if new_pubsub.subscribed and not self.subscribed:
4305 self.subscribed_event.set()
4306 self.health_check_response_counter = 0
4308 def on_slots_changed(self):
4309 # Observer hook invoked by NodesManager after a slots-cache refresh.
4310 # Schedule reconciliation on a dedicated worker thread so the caller
4311 # (typically MovedError handling in _execute_command or the topology
4312 # refresh thread in initialize()) is not blocked on the network I/O
4313 # performed by reinitialize_shard_subscriptions. Mirrors the async
4314 # path's asyncio.create_task model. No-op when there are no shard
4315 # subscriptions to reconcile.
4316 if not self.shard_channels:
4317 return
4318 # Takes no lock. This hook runs inline on whichever thread refreshed the
4319 # topology - including a command or pubsub thread that handled an
4320 # SMIGRATED push and still holds OSSMaintNotificationsHandler._lock.
4321 # Waiting here for a reconciliation pass that is itself waiting for a
4322 # per-node pubsub I/O lock held by a third thread blocked on that
4323 # handler lock is a deadlock cycle. The async counterpart takes no lock
4324 # here either (the event loop serializes it), so this is also the
4325 # shape sync/async parity asks for.
4326 self._submit_reconcile_work(self.reinitialize_shard_subscriptions)
4328 def _submit_reconcile_work(self, work: Callable[[], Any]) -> Optional["Future"]:
4329 """Run ``work`` on the reconciliation worker, without taking a lock.
4331 The executor is installed by ``reset()`` and never None afterwards, so
4332 there is nothing to create here and no creation race to serialize - see
4333 the ``_reconcile_executor`` comment in ``__init__`` for why the callers
4334 must not block. Only ``reset()`` swaps it, and a submit that loses that
4335 race raises ``RuntimeError``: read the attribute once so the swap cannot
4336 be observed half-done, and treat the rejection as "nothing to do", which
4337 it is - ``reset()`` has already dropped the shard channels this pass
4338 would have reconciled.
4339 """
4340 executor = self._reconcile_executor
4341 if executor is None:
4342 # Defensive: reset() runs from PubSub.__init__, so by the time any
4343 # caller can get here an executor is installed.
4344 return None
4345 try:
4346 future = executor.submit(work)
4347 except RuntimeError:
4348 return None
4349 # Consume the future's exception (if any) so it is not silently lost.
4350 # reinitialize_shard_subscriptions surfaces SlotNotCoveredError when a
4351 # slot is still transiently uncovered; route it through the same logger
4352 # channel as the async path for consistent observability.
4353 future.add_done_callback(self._log_reconcile_future_exception)
4354 return future
4356 @staticmethod
4357 def _log_reconcile_future_exception(future: "Future") -> None:
4358 if future.cancelled():
4359 return
4360 exc = future.exception()
4361 if exc is not None:
4362 logger.error(
4363 "shard subscription reconciliation failed: %r", exc, exc_info=exc
4364 )
4366 def reset(self) -> None:
4367 # Hold _shard_state_lock across the entire teardown so it observes
4368 # the same mutual-exclusion discipline as ssubscribe / sunsubscribe /
4369 # get_sharded_message / reinitialize_shard_subscriptions, which all
4370 # mutate shard_channels, _shard_channel_to_node, and
4371 # node_pubsub_mapping under this lock. Without it, super().reset()
4372 # rebinds shard_channels and pending_unsubscribe_shard_channels in
4373 # parallel with a concurrent user-thread mutation, silently dropping
4374 # subscription intent. The reconciliation executor is swapped out at the
4375 # end of that critical section; see the comment there for why it goes
4376 # last.
4377 with self._shard_state_lock:
4378 # Tear down per-node pubsubs (parity with async aclose) so they
4379 # don't leak their dedicated connections and don't replay stale
4380 # shard_channels via PubSub.on_connect on a subsequent reconnect.
4381 # Errors are swallowed because reset() is also a fallback path
4382 # from __del__; we cannot let one buggy per-node pubsub mask the
4383 # rest of the teardown.
4384 # The per-node I/O lock keeps the socket from being torn down
4385 # beneath a concurrent bounded poll parked in parse_response. It is
4386 # an RLock, so re-entry from this thread is fine; contention is with
4387 # another thread's bounded poll and is bounded by that poll's
4388 # timeout (an unbounded poll holds nullcontext() - see
4389 # _poll_io_lock). reset() is also the __del__ fallback path, where
4390 # the try below does not cover a blocking acquire - that boundedness
4391 # is what makes taking the lock here safe.
4392 for pubsub in self.node_pubsub_mapping.values():
4393 try:
4394 with self._pubsub_io_lock(pubsub):
4395 pubsub.reset()
4396 except Exception:
4397 pass
4398 # Drop the now-dead per-node pubsubs from the mapping so the
4399 # round-robin in _pubsubs_generator / _sharded_message_generator
4400 # cannot yield them between teardown and re-subscription.
4401 self.node_pubsub_mapping.clear()
4402 self._unreachable_nodes.clear()
4403 # Drop the throttle window too: a reused pubsub that keeps a
4404 # deadline armed before the teardown would skip the first repair
4405 # after resubscribing, delaying the move of its shard channels
4406 # off a node that is already gone.
4407 self._next_topology_repair = 0.0
4408 # _pubsubs_generator captures node_pubsub_mapping.values() into
4409 # a local list inside ``yield from``; clearing the mapping does
4410 # not reach references already held by that captured snapshot,
4411 # so a generator suspended mid-yield-from would still surface
4412 # the now-reset() per-node pubsubs after re-subscription.
4413 # Recreate it to drop the captured list. type(self) bypasses
4414 # the instance-level self-shadow established at __init__
4415 # (self._pubsubs_generator = self._pubsubs_generator()).
4416 self._pubsubs_generator = type(self)._pubsubs_generator(self)
4417 super().reset()
4418 self._shard_channel_to_node = {}
4419 # Swap in a fresh reconciliation executor and retire the old one.
4420 # Installing the replacement is what lets the scheduling sites stay
4421 # lock-free (see __init__); it costs nothing until something is
4422 # submitted, because a ThreadPoolExecutor spawns its worker on the
4423 # first submit(). Done last, after super().reset() has emptied
4424 # shard_channels, so a concurrent on_slots_changed either loses the
4425 # swap and is rejected or finds nothing left to reconcile.
4426 # cancel_futures drops queued reconciliation work; a pass already
4427 # running is serialized against us by _shard_state_lock, and
4428 # shutdown(wait=False) avoids waiting on the worker thread's join.
4429 retired = self._reconcile_executor
4430 self._reconcile_executor = ThreadPoolExecutor(
4431 max_workers=1,
4432 thread_name_prefix="redis-cluster-pubsub-reconcile",
4433 )
4434 if retired is not None:
4435 retired.shutdown(wait=False, cancel_futures=True)
4437 def get_redis_connection(self):
4438 """
4439 Get the Redis connection of the pubsub connected node.
4440 """
4441 if self.node is not None:
4442 return self.node.redis_connection
4444 def disconnect(self):
4445 """
4446 Disconnect the pubsub connection.
4447 """
4448 if self.connection:
4449 self.connection.disconnect()
4450 for pubsub in self.node_pubsub_mapping.values():
4451 if pubsub.connection:
4452 pubsub.connection.disconnect()
4455class ClusterPipeline(RedisCluster):
4456 """
4457 Support for Redis pipeline
4458 in cluster mode
4459 """
4461 ERRORS_ALLOW_RETRY = (
4462 ConnectionError,
4463 TimeoutError,
4464 MovedError,
4465 AskError,
4466 TryAgainError,
4467 )
4469 NO_SLOTS_COMMANDS = {"UNWATCH"}
4470 IMMEDIATE_EXECUTE_COMMANDS = {"WATCH", "UNWATCH"}
4471 UNWATCH_COMMANDS = {"DISCARD", "EXEC", "UNWATCH"}
4473 @deprecated_args(
4474 args_to_warn=[
4475 "cluster_error_retry_attempts",
4476 ],
4477 reason="Please configure the 'retry' object instead",
4478 version="6.0.0",
4479 )
4480 def __init__(
4481 self,
4482 nodes_manager: "NodesManager",
4483 commands_parser: "CommandsParser",
4484 result_callbacks: Optional[Dict[str, Callable]] = None,
4485 cluster_response_callbacks: Optional[Dict[str, Callable]] = None,
4486 startup_nodes: Optional[List["ClusterNode"]] = None,
4487 read_from_replicas: bool = False,
4488 load_balancing_strategy: Optional[LoadBalancingStrategy] = None,
4489 cluster_error_retry_attempts: int = DEFAULT_RETRY_COUNT,
4490 reinitialize_steps: int = 5,
4491 retry: Optional[Retry] = None,
4492 lock=None,
4493 transaction=False,
4494 policy_resolver: Optional[PolicyResolver] = None,
4495 event_dispatcher: Optional["EventDispatcher"] = None,
4496 metadata_resolver: Optional[MetadataResolver] = None,
4497 **kwargs,
4498 ):
4499 """ """
4500 self.command_stack = []
4501 self.nodes_manager = nodes_manager
4502 # Share the parent cluster's HIMPORT registry (held on the NodesManager and
4503 # referenced by every node pool). The inherited himport_prepare/discard/
4504 # discard_all mutate this one object, so a fieldset declared on the pipeline is
4505 # visible to the batched himport_set pre-flight exactly as on the parent client.
4506 self._himport_registry = nodes_manager.himport_registry
4507 self.commands_parser = commands_parser
4508 self.refresh_table_asap = False
4509 self.result_callbacks = (
4510 result_callbacks or self.__class__.RESULT_CALLBACKS.copy()
4511 )
4512 self.startup_nodes = startup_nodes if startup_nodes else []
4513 self.read_from_replicas = read_from_replicas
4514 self.load_balancing_strategy = load_balancing_strategy
4515 self.command_flags = self.__class__.COMMAND_FLAGS.copy()
4516 self.cluster_response_callbacks = cluster_response_callbacks
4517 self.reinitialize_counter = 0
4518 self.reinitialize_steps = reinitialize_steps
4519 if retry is not None:
4520 self.retry = retry
4521 else:
4522 self.retry = Retry(
4523 backoff=ExponentialWithJitterBackoff(
4524 base=DEFAULT_RETRY_BASE, cap=DEFAULT_RETRY_CAP
4525 ),
4526 retries=cluster_error_retry_attempts,
4527 )
4529 self.encoder = Encoder(
4530 kwargs.get("encoding", "utf-8"),
4531 kwargs.get("encoding_errors", "strict"),
4532 kwargs.get("decode_responses", False),
4533 )
4534 if lock is None:
4535 lock = threading.RLock()
4536 self._lock = lock
4537 self.parent_execute_command = super().execute_command
4538 self._execution_strategy: ExecutionStrategy = (
4539 PipelineStrategy(self) if not transaction else TransactionStrategy(self)
4540 )
4542 # For backward compatibility, mapping from existing policies to new one
4543 self._command_flags_mapping: dict[str, Union[RequestPolicy, ResponsePolicy]] = {
4544 self.__class__.RANDOM: RequestPolicy.DEFAULT_KEYLESS,
4545 self.__class__.PRIMARIES: RequestPolicy.ALL_SHARDS,
4546 self.__class__.ALL_NODES: RequestPolicy.ALL_NODES,
4547 self.__class__.REPLICAS: RequestPolicy.ALL_REPLICAS,
4548 self.__class__.DEFAULT_NODE: RequestPolicy.DEFAULT_NODE,
4549 SLOT_ID: RequestPolicy.DEFAULT_KEYED,
4550 }
4552 self._policies_callback_mapping: dict[
4553 Union[RequestPolicy, ResponsePolicy], Callable
4554 ] = {
4555 RequestPolicy.DEFAULT_KEYLESS: lambda command_name: [
4556 self.get_keyless_target_node(command_name)
4557 ],
4558 RequestPolicy.DEFAULT_KEYED: lambda command,
4559 *args: self.get_nodes_from_slot(command, *args),
4560 RequestPolicy.DEFAULT_NODE: lambda: [self.get_default_node()],
4561 RequestPolicy.ALL_SHARDS: self.get_primaries,
4562 RequestPolicy.ALL_NODES: self.get_nodes,
4563 RequestPolicy.ALL_REPLICAS: self.get_replicas,
4564 RequestPolicy.MULTI_SHARD: lambda *args,
4565 **kwargs: self._split_multi_shard_command(*args, **kwargs),
4566 RequestPolicy.SPECIAL: self.get_special_nodes,
4567 ResponsePolicy.DEFAULT_KEYLESS: lambda res: res,
4568 ResponsePolicy.DEFAULT_KEYED: lambda res: res,
4569 }
4571 # ``RedisCluster.pipeline`` passes the client's own resolvers, so a pipeline routes by
4572 # whatever the client routes by. Only a pipeline built directly, without either, falls
4573 # back to the static default. Precedence between the two mirrors ``RedisCluster`` -
4574 # see the note there.
4575 if metadata_resolver is None:
4576 self._metadata_resolver: MetadataResolver = StaticMetadataResolver()
4577 else:
4578 self._metadata_resolver = metadata_resolver
4580 if policy_resolver is None:
4581 self._policy_resolver: PolicyResolver = StaticPolicyResolver(
4582 metadata_resolver=self._metadata_resolver
4583 )
4584 else:
4585 self._policy_resolver = policy_resolver
4587 if event_dispatcher is None:
4588 self._event_dispatcher = EventDispatcher()
4589 else:
4590 self._event_dispatcher = event_dispatcher
4592 def __repr__(self):
4593 """ """
4594 return f"{type(self).__name__}"
4596 def __enter__(self):
4597 """ """
4598 return self
4600 def __exit__(self, exc_type, exc_value, traceback):
4601 """ """
4602 self.reset()
4604 def __del__(self):
4605 try:
4606 self.reset()
4607 except Exception:
4608 pass
4610 def __len__(self):
4611 """ """
4612 return len(self._execution_strategy.command_queue)
4614 def __bool__(self):
4615 "Pipeline instances should always evaluate to True on Python 3+"
4616 return True
4618 def execute_command(self, *args, **kwargs):
4619 """
4620 Wrapper function for pipeline_execute_command
4621 """
4622 return self._execution_strategy.execute_command(*args, **kwargs)
4624 def pipeline_execute_command(self, *args, **options):
4625 """
4626 Stage a command to be executed when execute() is next called
4628 Returns the current Pipeline object back so commands can be
4629 chained together, such as:
4631 pipe = pipe.set('foo', 'bar').incr('baz').decr('bang')
4633 At some other point, you can then run: pipe.execute(),
4634 which will execute all commands queued in the pipe.
4635 """
4636 return self._execution_strategy.execute_command(*args, **options)
4638 def annotate_exception(self, exception, number, command):
4639 """
4640 Provides extra context to the exception prior to it being handled
4641 """
4642 self._execution_strategy.annotate_exception(exception, number, command)
4644 def execute(self, raise_on_error: bool = True) -> List[Any]:
4645 """
4646 Execute all the commands in the current pipeline
4647 """
4649 try:
4650 return self._execution_strategy.execute(raise_on_error)
4651 finally:
4652 self.reset()
4654 def reset(self):
4655 """
4656 Reset back to empty pipeline.
4657 """
4658 self._execution_strategy.reset()
4660 def send_cluster_commands(
4661 self, stack, raise_on_error=True, allow_redirections=True
4662 ):
4663 return self._execution_strategy.send_cluster_commands(
4664 stack, raise_on_error=raise_on_error, allow_redirections=allow_redirections
4665 )
4667 def exists(self, *keys):
4668 return self._execution_strategy.exists(*keys)
4670 def eval(self):
4671 """ """
4672 return self._execution_strategy.eval()
4674 def multi(self):
4675 """
4676 Start a transactional block of the pipeline after WATCH commands
4677 are issued. End the transactional block with `execute`.
4678 """
4679 self._execution_strategy.multi()
4681 def load_scripts(self):
4682 """ """
4683 self._execution_strategy.load_scripts()
4685 def discard(self):
4686 """ """
4687 self._execution_strategy.discard()
4689 def watch(self, *names):
4690 """Watches the values at keys ``names``"""
4691 self._execution_strategy.watch(*names)
4693 def unwatch(self):
4694 """Unwatches all previously specified keys"""
4695 self._execution_strategy.unwatch()
4697 def script_load_for_pipeline(self, *args, **kwargs):
4698 self._execution_strategy.script_load_for_pipeline(*args, **kwargs)
4700 def delete(self, *names):
4701 self._execution_strategy.delete(*names)
4703 def unlink(self, *names):
4704 self._execution_strategy.unlink(*names)
4707def block_pipeline_command(name: str) -> Callable[..., Any]:
4708 """
4709 Prints error because some pipelined commands should
4710 be blocked when running in cluster-mode
4711 """
4713 def inner(*args, **kwargs):
4714 raise RedisClusterException(
4715 f"ERROR: Calling pipelined function {name} is blocked "
4716 f"when running redis in cluster mode..."
4717 )
4719 return inner
4722def is_zero_key_eval_command(*args) -> bool:
4723 """
4724 True for EVAL/EVALSHA with numkeys=0 (any primary).
4725 """
4726 if len(args) < 3:
4727 return False
4728 if str(args[0]).upper() not in ("EVAL", "EVALSHA"):
4729 return False
4730 try:
4731 return int(args[2]) == 0
4732 except (TypeError, ValueError):
4733 return False
4736# Blocked pipeline commands
4737PIPELINE_BLOCKED_COMMANDS = (
4738 "BGREWRITEAOF",
4739 "BGSAVE",
4740 "BITOP",
4741 "BRPOPLPUSH",
4742 "CLIENT GETNAME",
4743 "CLIENT KILL",
4744 "CLIENT LIST",
4745 "CLIENT SETNAME",
4746 "CLIENT",
4747 "CONFIG GET",
4748 "CONFIG RESETSTAT",
4749 "CONFIG REWRITE",
4750 "CONFIG SET",
4751 "CONFIG",
4752 "DBSIZE",
4753 "ECHO",
4754 "FLUSHALL",
4755 "FLUSHDB",
4756 "INFO",
4757 "KEYS",
4758 "LASTSAVE",
4759 "MGET",
4760 "MGET NONATOMIC",
4761 "MOVE",
4762 "MSET",
4763 "MSETEX",
4764 "MSET NONATOMIC",
4765 "MSETNX",
4766 "PFCOUNT",
4767 "PFMERGE",
4768 "PING",
4769 "PUBLISH",
4770 "RANDOMKEY",
4771 "READONLY",
4772 "READWRITE",
4773 "RENAME",
4774 "RENAMENX",
4775 "RPOPLPUSH",
4776 "SAVE",
4777 "SCAN",
4778 "SCRIPT EXISTS",
4779 "SCRIPT FLUSH",
4780 "SCRIPT KILL",
4781 "SCRIPT LOAD",
4782 "SCRIPT",
4783 "SDIFF",
4784 "SDIFFSTORE",
4785 "SENTINEL GET MASTER ADDR BY NAME",
4786 "SENTINEL MASTER",
4787 "SENTINEL MASTERS",
4788 "SENTINEL MONITOR",
4789 "SENTINEL REMOVE",
4790 "SENTINEL SENTINELS",
4791 "SENTINEL SET",
4792 "SENTINEL SLAVES",
4793 "SENTINEL",
4794 "SHUTDOWN",
4795 "SINTER",
4796 "SINTERSTORE",
4797 "SLAVEOF",
4798 "SLOWLOG GET",
4799 "SLOWLOG LEN",
4800 "SLOWLOG RESET",
4801 "SLOWLOG",
4802 "SMOVE",
4803 "SORT",
4804 "SUNION",
4805 "SUNIONSTORE",
4806 "TIME",
4807)
4808for command in PIPELINE_BLOCKED_COMMANDS:
4809 command = command.replace(" ", "_").lower()
4811 setattr(ClusterPipeline, command, block_pipeline_command(command))
4813# client_list_iter has no wire command of its own to add to
4814# PIPELINE_BLOCKED_COMMANDS - it sends CLIENT LIST, blocked above under its
4815# own name - so block it explicitly here too, or it would fall through to
4816# the inherited implementation and queue CLIENT LIST like a real pipelined
4817# command instead of raising.
4818setattr(ClusterPipeline, "client_list_iter", block_pipeline_command("client_list_iter"))
4821class PipelineCommand:
4822 """ """
4824 def __init__(self, args, options=None, position=None):
4825 self.args = args
4826 if options is None:
4827 options = {}
4828 self.options = options
4829 self.position = position
4830 self.result = None
4831 self.node = None
4832 self.asking = False
4833 # Either record type: a policy resolver serves ``CommandPolicies``, while the
4834 # fallbacks below reuse the shared ``CommandMetadata`` defaults. Only the two routing
4835 # policies, which both carry, are ever read.
4836 self.command_policies: Optional[Union[CommandPolicies, CommandMetadata]] = None
4839class NodeCommands:
4840 """ """
4842 def __init__(
4843 self, parse_response, connection_pool: ConnectionPool, connection: Connection
4844 ):
4845 """ """
4846 self.parse_response = parse_response
4847 self.connection_pool = connection_pool
4848 self.connection = connection
4849 self.commands = []
4851 def append(self, c):
4852 """ """
4853 self.commands.append(c)
4855 def write(self):
4856 """
4857 Code borrowed from Redis so it can be fixed
4858 """
4859 connection = self.connection
4860 commands = self.commands
4862 # We are going to clobber the commands with the write, so go ahead
4863 # and ensure that nothing is sitting there from a previous run.
4864 for c in commands:
4865 c.result = None
4867 # build up all commands into a single request to increase network perf
4868 # send all the commands and catch connection and timeout errors.
4869 try:
4870 connection.send_packed_command(
4871 connection.pack_commands([c.args for c in commands])
4872 )
4873 except (ConnectionError, TimeoutError) as e:
4874 for c in commands:
4875 c.result = e
4877 def read(self):
4878 """ """
4879 connection = self.connection
4880 for c in self.commands:
4881 # if there is a result on this command,
4882 # it means we ran into an exception
4883 # like a connection error. Trying to parse
4884 # a response on a connection that
4885 # is no longer open will result in a
4886 # connection error raised by redis-py.
4887 # but redis-py doesn't check in parse_response
4888 # that the sock object is
4889 # still set and if you try to
4890 # read from a closed connection, it will
4891 # result in an AttributeError because
4892 # it will do a readline() call on None.
4893 # This can have all kinds of nasty side-effects.
4894 # Treating this case as a connection error
4895 # is fine because it will dump
4896 # the connection object back into the
4897 # pool and on the next write, it will
4898 # explicitly open the connection and all will be well.
4899 if c.result is None:
4900 try:
4901 c.result = self.parse_response(connection, c.args[0], **c.options)
4902 except (ConnectionError, TimeoutError) as e:
4903 for c in self.commands:
4904 c.result = e
4905 return
4906 except RedisError:
4907 c.result = sys.exc_info()[1]
4910class ExecutionStrategy(ABC):
4911 @property
4912 @abstractmethod
4913 def command_queue(self):
4914 pass
4916 @abstractmethod
4917 def execute_command(self, *args, **kwargs):
4918 """
4919 Execution flow for current execution strategy.
4921 See: ClusterPipeline.execute_command()
4922 """
4923 pass
4925 @abstractmethod
4926 def annotate_exception(self, exception, number, command):
4927 """
4928 Annotate exception according to current execution strategy.
4930 See: ClusterPipeline.annotate_exception()
4931 """
4932 pass
4934 @abstractmethod
4935 def pipeline_execute_command(self, *args, **options):
4936 """
4937 Pipeline execution flow for current execution strategy.
4939 See: ClusterPipeline.pipeline_execute_command()
4940 """
4941 pass
4943 @abstractmethod
4944 def execute(self, raise_on_error: bool = True) -> List[Any]:
4945 """
4946 Executes current execution strategy.
4948 See: ClusterPipeline.execute()
4949 """
4950 pass
4952 @abstractmethod
4953 def send_cluster_commands(
4954 self, stack, raise_on_error=True, allow_redirections=True
4955 ):
4956 """
4957 Sends commands according to current execution strategy.
4959 See: ClusterPipeline.send_cluster_commands()
4960 """
4961 pass
4963 @abstractmethod
4964 def reset(self):
4965 """
4966 Resets current execution strategy.
4968 See: ClusterPipeline.reset()
4969 """
4970 pass
4972 @abstractmethod
4973 def exists(self, *keys):
4974 pass
4976 @abstractmethod
4977 def eval(self):
4978 pass
4980 @abstractmethod
4981 def multi(self):
4982 """
4983 Starts transactional context.
4985 See: ClusterPipeline.multi()
4986 """
4987 pass
4989 @abstractmethod
4990 def load_scripts(self):
4991 pass
4993 @abstractmethod
4994 def watch(self, *names):
4995 pass
4997 @abstractmethod
4998 def unwatch(self):
4999 """
5000 Unwatches all previously specified keys
5002 See: ClusterPipeline.unwatch()
5003 """
5004 pass
5006 @abstractmethod
5007 def script_load_for_pipeline(self, *args, **kwargs):
5008 pass
5010 @abstractmethod
5011 def delete(self, *names):
5012 """
5013 "Delete a key specified by ``names``"
5015 See: ClusterPipeline.delete()
5016 """
5017 pass
5019 @abstractmethod
5020 def unlink(self, *names):
5021 """
5022 "Unlink a key specified by ``names``"
5024 See: ClusterPipeline.unlink()
5025 """
5026 pass
5028 @abstractmethod
5029 def discard(self):
5030 pass
5033class AbstractStrategy(ExecutionStrategy):
5034 def __init__(
5035 self,
5036 pipe: ClusterPipeline,
5037 ):
5038 self._command_queue: List[PipelineCommand] = []
5039 self._pipe = pipe
5040 self._nodes_manager = self._pipe.nodes_manager
5042 @property
5043 def command_queue(self):
5044 return self._command_queue
5046 @command_queue.setter
5047 def command_queue(self, queue: List[PipelineCommand]):
5048 self._command_queue = queue
5050 @abstractmethod
5051 def execute_command(self, *args, **kwargs):
5052 pass
5054 def pipeline_execute_command(self, *args, **options):
5055 self._command_queue.append(
5056 PipelineCommand(args, options, len(self._command_queue))
5057 )
5058 return self._pipe
5060 def _himport_prepare_pipeline(self, redis_node, conn, commands):
5061 """Delegate to the shared sync HIMPORT executor."""
5062 _himport_exec.prepare_pipeline(redis_node, conn, [args for args, _ in commands])
5064 @abstractmethod
5065 def execute(self, raise_on_error: bool = True) -> List[Any]:
5066 pass
5068 @abstractmethod
5069 def send_cluster_commands(
5070 self, stack, raise_on_error=True, allow_redirections=True
5071 ):
5072 pass
5074 @abstractmethod
5075 def reset(self):
5076 pass
5078 def exists(self, *keys):
5079 return self.execute_command("EXISTS", *keys)
5081 def eval(self):
5082 """ """
5083 raise RedisClusterException("method eval() is not implemented")
5085 def load_scripts(self):
5086 """ """
5087 raise RedisClusterException("method load_scripts() is not implemented")
5089 def script_load_for_pipeline(self, *args, **kwargs):
5090 """ """
5091 raise RedisClusterException(
5092 "method script_load_for_pipeline() is not implemented"
5093 )
5095 def annotate_exception(self, exception, number, command):
5096 """
5097 Provides extra context to the exception prior to it being handled
5098 """
5099 cmd = " ".join(map(safe_str, command))
5100 msg = (
5101 f"Command # {number} ({truncate_text(cmd)}) of pipeline "
5102 f"caused error: {exception.args[0]}"
5103 )
5104 exception.args = (msg,) + exception.args[1:]
5107class PipelineStrategy(AbstractStrategy):
5108 def __init__(self, pipe: ClusterPipeline):
5109 super().__init__(pipe)
5110 self.command_flags = pipe.command_flags
5112 def execute_command(self, *args, **kwargs):
5113 return self.pipeline_execute_command(*args, **kwargs)
5115 def _raise_first_error(self, stack, start_time):
5116 """
5117 Raise the first exception on the stack
5118 """
5119 for c in stack:
5120 r = c.result
5121 if isinstance(r, Exception):
5122 self.annotate_exception(r, c.position + 1, c.args)
5124 record_operation_duration(
5125 command_name="PIPELINE",
5126 duration_seconds=time.monotonic() - start_time,
5127 error=r,
5128 )
5130 raise r
5132 def execute(self, raise_on_error: bool = True) -> List[Any]:
5133 stack = self._command_queue
5134 if not stack:
5135 return []
5137 try:
5138 return self.send_cluster_commands(stack, raise_on_error)
5139 finally:
5140 self.reset()
5142 def reset(self):
5143 """
5144 Reset back to empty pipeline.
5145 """
5146 self._command_queue = []
5148 def send_cluster_commands(
5149 self, stack, raise_on_error=True, allow_redirections=True
5150 ):
5151 """
5152 Wrapper for RedisCluster.ERRORS_ALLOW_RETRY errors handling.
5154 If one of the retryable exceptions has been thrown we assume that:
5155 - connection_pool was disconnected
5156 - connection_pool was reset
5157 - refresh_table_asap set to True
5159 It will try the number of times specified by
5160 the retries in config option "self.retry"
5161 which defaults to 10 unless manually configured.
5163 If it reaches the number of times, the command will
5164 raises ClusterDownException.
5165 """
5166 if not stack:
5167 return []
5168 retry_attempts = self._pipe.retry.get_retries()
5169 while True:
5170 try:
5171 return self._send_cluster_commands(
5172 stack,
5173 raise_on_error=raise_on_error,
5174 allow_redirections=allow_redirections,
5175 )
5176 except RedisCluster.ERRORS_ALLOW_RETRY as e:
5177 if retry_attempts > 0:
5178 # Try again with the new cluster setup. All other errors
5179 # should be raised.
5180 retry_attempts -= 1
5181 pass
5182 else:
5183 raise e
5185 def _send_cluster_commands(
5186 self, stack, raise_on_error=True, allow_redirections=True
5187 ):
5188 """
5189 Send a bunch of cluster commands to the redis cluster.
5191 `allow_redirections` If the pipeline should follow
5192 `ASK` & `MOVED` responses automatically. If set
5193 to false it will raise RedisClusterException.
5194 """
5195 # the first time sending the commands we send all of
5196 # the commands that were queued up.
5197 # if we have to run through it again, we only retry
5198 # the commands that failed.
5199 attempt = sorted(stack, key=lambda x: x.position)
5200 is_default_node = False
5201 # build a list of node objects based on node names we need to
5202 nodes: dict[str, NodeCommands] = {}
5203 # node objects keyed by name, so each node's connection can be pre-flighted
5204 # for HIMPORT SET (PREPARE) before the batched write.
5205 node_objs: dict = {}
5206 nodes_written = 0
5207 nodes_read = 0
5209 try:
5210 # as we move through each command that still needs to be processed,
5211 # we figure out the slot number that command maps to, then from
5212 # the slot determine the node.
5213 for c in attempt:
5214 command_policies = self._pipe._policy_resolver.resolve(
5215 c.args[0].lower()
5216 )
5217 # refer to our internal node -> slot table that
5218 # tells us where a given command should route to.
5219 # (it might be possible we have a cached node that no longer
5220 # exists in the cluster, which is why we do this in a loop)
5221 passed_targets = c.options.pop("target_nodes", None)
5222 if passed_targets and not self._is_nodes_flag(passed_targets):
5223 target_nodes = self._parse_target_nodes(passed_targets)
5225 if not command_policies:
5226 command_policies = _DEFAULT_KEYLESS_METADATA
5227 else:
5228 if not command_policies:
5229 command = c.args[0].upper()
5230 if (
5231 len(c.args) >= 2
5232 and f"{c.args[0]} {c.args[1]}".upper()
5233 in self._pipe.command_flags
5234 ):
5235 command = f"{c.args[0]} {c.args[1]}".upper()
5237 # We only could resolve key properties if command is not
5238 # in a list of pre-defined request policies
5239 command_flag = self.command_flags.get(command)
5240 if not command_flag:
5241 # Fallback to default policy.
5242 # EVAL/EVALSHA must not use _get_command_keys(): Redis
5243 # <7 breaks on COMMAND GETKEYS when numkeys is 0.
5244 # Other unflagged commands keep the keyless fallback.
5245 if command in ("EVAL", "EVALSHA"):
5246 command_policies = _DEFAULT_KEYED_METADATA
5247 else:
5248 if not self._pipe.get_default_node():
5249 keys = None
5250 else:
5251 keys = self._pipe._get_command_keys(*c.args)
5252 if not keys or len(keys) == 0:
5253 command_policies = _DEFAULT_KEYLESS_METADATA
5254 else:
5255 command_policies = _DEFAULT_KEYED_METADATA
5256 else:
5257 if command_flag in self._pipe._command_flags_mapping:
5258 command_policies = _METADATA_BY_REQUEST_POLICY[
5259 self._pipe._command_flags_mapping[command_flag]
5260 ]
5261 else:
5262 command_policies = _DEFAULT_KEYLESS_METADATA
5264 target_nodes = self._determine_nodes(
5265 *c.args,
5266 request_policy=command_policies.request_policy,
5267 node_flag=passed_targets,
5268 )
5269 if not target_nodes:
5270 raise RedisClusterException(
5271 f"No targets were found to execute {c.args} command on"
5272 )
5273 c.command_policies = command_policies
5274 if len(target_nodes) > 1:
5275 raise RedisClusterException(
5276 f"Too many targets for command {c.args}"
5277 )
5279 node = target_nodes[0]
5280 if node == self._pipe.get_default_node():
5281 is_default_node = True
5283 # now that we know the name of the node
5284 # ( it's just a string in the form of host:port )
5285 # we can build a list of commands for each node.
5286 node_name = node.name
5287 if node_name not in nodes:
5288 redis_node = self._pipe.get_redis_connection(node)
5289 try:
5290 connection = get_connection(redis_node)
5291 except (ConnectionError, TimeoutError):
5292 # Release any connections we've already acquired before clearing nodes
5293 for n in nodes.values():
5294 n.connection_pool.release(n.connection)
5295 # Connection retries are being handled in the node's
5296 # Retry object. Reinitialize the node -> slot table.
5297 self._nodes_manager.initialize()
5298 if is_default_node:
5299 self._pipe.replace_default_node()
5300 nodes = {}
5301 raise
5302 nodes[node_name] = NodeCommands(
5303 redis_node.parse_response,
5304 redis_node.connection_pool,
5305 connection,
5306 )
5307 node_objs[node_name] = node
5308 nodes[node_name].append(c)
5310 # send the commands in sequence.
5311 # we write to all the open sockets for each node first,
5312 # before reading anything
5313 # this allows us to flush all the requests out across the
5314 # network
5315 # so that we can read them from different sockets as they come back.
5316 # we don't multiplex on the sockets as they come available,
5317 # but that shouldn't make too much difference.
5319 # HIMPORT SETs in the batch need their fieldsets prepared on each
5320 # node's connection first; the packed write bypasses the per-command
5321 # lazy prepare, so pre-flight the PREPARE (once per node) here.
5322 for node_name, n in nodes.items():
5323 redis_node = self._pipe.get_redis_connection(node_objs[node_name])
5324 self._himport_prepare_pipeline(
5325 redis_node, n.connection, [(c.args, c.options) for c in n.commands]
5326 )
5328 # Start timing for observability
5329 start_time = time.monotonic()
5331 node_commands = nodes.values()
5332 for n in node_commands:
5333 nodes_written += 1
5334 n.write()
5336 for n in node_commands:
5337 n.read()
5339 # Find the first error in this node's commands, if any
5340 node_error = None
5341 for cmd in n.commands:
5342 if isinstance(cmd.result, Exception):
5343 node_error = cmd.result
5344 break
5346 record_operation_duration(
5347 command_name="PIPELINE",
5348 duration_seconds=time.monotonic() - start_time,
5349 server_address=n.connection.host,
5350 server_port=n.connection.port,
5351 db_namespace=str(n.connection.db),
5352 error=node_error,
5353 )
5354 nodes_read += 1
5355 finally:
5356 # release all the redis connections we allocated earlier
5357 # back into the connection pool.
5358 # if the connection is dirty (that is: we've written
5359 # commands to it, but haven't read the responses), we need
5360 # to close the connection before returning it to the pool.
5361 # otherwise, the next caller to use this connection will
5362 # read the response from _this_ request, not its own request.
5363 # disconnecting discards the dirty state & forces the next
5364 # caller to reconnect.
5365 # NOTE: dicts have a consistent ordering; we're iterating
5366 # through nodes.values() in the same order as we are when
5367 # reading / writing to the connections above, which is critical
5368 # for how we're using the nodes_written/nodes_read offsets.
5369 for i, n in enumerate(nodes.values()):
5370 if i < nodes_written and i >= nodes_read:
5371 n.connection.disconnect()
5372 n.connection_pool.release(n.connection)
5374 # if the response isn't an exception it is a
5375 # valid response from the node
5376 # we're all done with that command, YAY!
5377 # if we have more commands to attempt, we've run into problems.
5378 # collect all the commands we are allowed to retry.
5379 # (MOVED, ASK, or connection errors or timeout errors)
5380 attempt = sorted(
5381 (
5382 c
5383 for c in attempt
5384 if isinstance(c.result, ClusterPipeline.ERRORS_ALLOW_RETRY)
5385 ),
5386 key=lambda x: x.position,
5387 )
5388 if attempt and allow_redirections:
5389 # RETRY MAGIC HAPPENS HERE!
5390 # send these remaining commands one at a time using `execute_command`
5391 # in the main client. This keeps our retry logic
5392 # in one place mostly,
5393 # and allows us to be more confident in correctness of behavior.
5394 # at this point any speed gains from pipelining have been lost
5395 # anyway, so we might as well make the best
5396 # attempt to get the correct behavior.
5397 #
5398 # The client command will handle retries for each
5399 # individual command sequentially as we pass each
5400 # one into `execute_command`. Any exceptions
5401 # that bubble out should only appear once all
5402 # retries have been exhausted.
5403 #
5404 # If a lot of commands have failed, we'll be setting the
5405 # flag to rebuild the slots table from scratch.
5406 # So MOVED errors should correct themselves fairly quickly.
5407 self._pipe.reinitialize_counter += 1
5408 if self._pipe._should_reinitialized():
5409 self._nodes_manager.initialize()
5410 if is_default_node:
5411 self._pipe.replace_default_node()
5412 for c in attempt:
5413 try:
5414 # send each command individually like we
5415 # do in the main client.
5416 c.result = self._pipe.parent_execute_command(*c.args, **c.options)
5417 except RedisError as e:
5418 c.result = e
5420 # turn the response back into a simple flat array that corresponds
5421 # to the sequence of commands issued in the stack in pipeline.execute()
5422 response = []
5423 for c in sorted(stack, key=lambda x: x.position):
5424 if c.args[0] in self._pipe.cluster_response_callbacks:
5425 # Remove keys entry, it needs only for cache.
5426 c.options.pop("keys", None)
5427 c.result = self._pipe._policies_callback_mapping[
5428 c.command_policies.response_policy
5429 ](
5430 self._pipe.cluster_response_callbacks[c.args[0]](
5431 c.result, **c.options
5432 )
5433 )
5434 response.append(c.result)
5436 if raise_on_error:
5437 self._raise_first_error(stack, start_time)
5439 return response
5441 def _is_nodes_flag(self, target_nodes):
5442 return isinstance(target_nodes, str) and target_nodes in self._pipe.node_flags
5444 def _parse_target_nodes(self, target_nodes):
5445 if isinstance(target_nodes, list):
5446 nodes = target_nodes
5447 elif isinstance(target_nodes, ClusterNode):
5448 # Supports passing a single ClusterNode as a variable
5449 nodes = [target_nodes]
5450 elif isinstance(target_nodes, dict):
5451 # Supports dictionaries of the format {node_name: node}.
5452 # It enables to execute commands with multi nodes as follows:
5453 # rc.cluster_save_config(rc.get_primaries())
5454 nodes = target_nodes.values()
5455 else:
5456 raise TypeError(
5457 "target_nodes type can be one of the following: "
5458 "node_flag (PRIMARIES, REPLICAS, RANDOM, ALL_NODES),"
5459 "ClusterNode, list<ClusterNode>, or dict<any, ClusterNode>. "
5460 f"The passed type is {type(target_nodes)}"
5461 )
5462 return nodes
5464 def _determine_nodes(
5465 self, *args, request_policy: Optional[RequestPolicy] = None, **kwargs
5466 ) -> List["ClusterNode"]:
5467 # Determine which nodes should be executed the command on.
5468 # Returns a list of target nodes.
5469 command = args[0].upper()
5470 if (
5471 len(args) >= 2
5472 and f"{args[0]} {args[1]}".upper() in self._pipe.command_flags
5473 ):
5474 command = f"{args[0]} {args[1]}".upper()
5476 # The caller resolves the command's own policy - see
5477 # ``RedisCluster._resolve_command_policies`` - so the only decision left here is
5478 # an explicit nodes flag, which overrides it.
5479 nodes_flag = kwargs.pop("nodes_flag", None)
5480 if nodes_flag and self._is_nodes_flag(nodes_flag):
5481 # nodes flag passed by the user
5482 if nodes_flag in self._pipe._command_flags_mapping:
5483 request_policy = self._pipe._command_flags_mapping[nodes_flag]
5485 if request_policy is None:
5486 raise RedisClusterException(
5487 f"No targets were found to execute {args} command on"
5488 )
5490 policy_callback = self._pipe._policies_callback_mapping[request_policy]
5492 if request_policy == RequestPolicy.DEFAULT_KEYED:
5493 nodes = policy_callback(command, *args)
5494 elif request_policy == RequestPolicy.MULTI_SHARD:
5495 nodes = policy_callback(*args, **kwargs)
5496 elif request_policy == RequestPolicy.DEFAULT_KEYLESS:
5497 nodes = policy_callback(args[0])
5498 else:
5499 nodes = policy_callback()
5501 if args[0].lower() == "ft.aggregate":
5502 self._aggregate_nodes = nodes
5504 return nodes
5506 def multi(self):
5507 raise RedisClusterException(
5508 "method multi() is not supported outside of transactional context"
5509 )
5511 def discard(self):
5512 raise RedisClusterException(
5513 "method discard() is not supported outside of transactional context"
5514 )
5516 def watch(self, *names):
5517 raise RedisClusterException(
5518 "method watch() is not supported outside of transactional context"
5519 )
5521 def unwatch(self, *names):
5522 raise RedisClusterException(
5523 "method unwatch() is not supported outside of transactional context"
5524 )
5526 def delete(self, *names):
5527 if len(names) != 1:
5528 raise RedisClusterException(
5529 "deleting multiple keys is not implemented in pipeline command"
5530 )
5532 return self.execute_command("DEL", names[0])
5534 def unlink(self, *names):
5535 if len(names) != 1:
5536 raise RedisClusterException(
5537 "unlinking multiple keys is not implemented in pipeline command"
5538 )
5540 return self.execute_command("UNLINK", names[0])
5543class TransactionStrategy(AbstractStrategy):
5544 NO_SLOTS_COMMANDS = {"UNWATCH"}
5545 IMMEDIATE_EXECUTE_COMMANDS = {"WATCH", "UNWATCH"}
5546 UNWATCH_COMMANDS = {"DISCARD", "EXEC", "UNWATCH"}
5547 SLOT_REDIRECT_ERRORS = (AskError, MovedError)
5548 CONNECTION_ERRORS = (
5549 ConnectionError,
5550 OSError,
5551 ClusterDownError,
5552 SlotNotCoveredError,
5553 )
5555 def __init__(self, pipe: ClusterPipeline):
5556 super().__init__(pipe)
5557 self._explicit_transaction = False
5558 self._watching = False
5559 self._pipeline_slots: Set[int] = set()
5560 # True once a keyed (non-slot-agnostic) command has fixed the slot
5561 self._transaction_has_keyed_slot = False
5562 self._transaction_connection: Optional[Connection] = None
5563 self._executing = False
5564 self._retry = copy(self._pipe.retry)
5565 self._retry.update_supported_errors(
5566 RedisCluster.ERRORS_ALLOW_RETRY + self.SLOT_REDIRECT_ERRORS
5567 )
5569 def _resolve_transaction_slot(self, *args) -> Optional[int]:
5570 """
5571 Pick a slot for a transactional pipeline command.
5573 Zero-key EVAL/EVALSHA can run on any primary. Reuse an existing
5574 transaction slot when present so multiple zero-key scripts (or a
5575 mix with keyed commands) stay single-slot.
5576 """
5577 if args[0] in ClusterPipeline.NO_SLOTS_COMMANDS:
5578 return None
5580 if is_zero_key_eval_command(*args):
5581 if self._pipeline_slots:
5582 return next(iter(self._pipeline_slots))
5583 return self._pipe.determine_slot(*args)
5585 slot_number = self._pipe.determine_slot(*args)
5586 if (
5587 slot_number is not None
5588 and self._pipeline_slots
5589 and slot_number not in self._pipeline_slots
5590 and not self._transaction_has_keyed_slot
5591 ):
5592 # Prior slots came only from zero-key scripts; retarget.
5593 self._pipeline_slots.clear()
5594 if slot_number is not None:
5595 self._transaction_has_keyed_slot = True
5596 return slot_number
5598 def _get_client_and_connection_for_transaction(self) -> Tuple[Redis, Connection]:
5599 """
5600 Find a connection for a pipeline transaction.
5602 For running an atomic transaction, watch keys ensure that contents have not been
5603 altered as long as the watch commands for those keys were sent over the same
5604 connection. So once we start watching a key, we fetch a connection to the
5605 node that owns that slot and reuse it.
5606 """
5607 if not self._pipeline_slots:
5608 raise RedisClusterException(
5609 "At least a command with a key is needed to identify a node"
5610 )
5612 node: ClusterNode = self._nodes_manager.get_node_from_slot(
5613 list(self._pipeline_slots)[0], False
5614 )
5615 redis_node: Redis = self._pipe.get_redis_connection(node)
5616 if self._transaction_connection:
5617 if not redis_node.connection_pool.owns_connection(
5618 self._transaction_connection
5619 ):
5620 previous_node = self._nodes_manager.find_connection_owner(
5621 self._transaction_connection
5622 )
5623 previous_node.connection_pool.release(self._transaction_connection)
5624 self._transaction_connection = None
5626 if not self._transaction_connection:
5627 self._transaction_connection = get_connection(redis_node)
5629 return redis_node, self._transaction_connection
5631 def execute_command(self, *args, **kwargs):
5632 slot_number: Optional[int] = None
5633 if args[0] not in ClusterPipeline.NO_SLOTS_COMMANDS:
5634 slot_number = self._resolve_transaction_slot(*args)
5636 if (
5637 self._watching or args[0] in self.IMMEDIATE_EXECUTE_COMMANDS
5638 ) and not self._explicit_transaction:
5639 if args[0] == "WATCH":
5640 self._validate_watch()
5642 if slot_number is not None:
5643 if self._pipeline_slots and slot_number not in self._pipeline_slots:
5644 raise CrossSlotTransactionError(
5645 "Cannot watch or send commands on different slots"
5646 )
5648 self._pipeline_slots.add(slot_number)
5649 elif args[0] not in self.NO_SLOTS_COMMANDS:
5650 raise RedisClusterException(
5651 f"Cannot identify slot number for command: {args[0]},"
5652 "it cannot be triggered in a transaction"
5653 )
5655 return self._immediate_execute_command(*args, **kwargs)
5656 else:
5657 if slot_number is not None:
5658 self._pipeline_slots.add(slot_number)
5660 return self.pipeline_execute_command(*args, **kwargs)
5662 def _validate_watch(self):
5663 if self._explicit_transaction:
5664 raise RedisError("Cannot issue a WATCH after a MULTI")
5666 self._watching = True
5668 def _immediate_execute_command(self, *args, **options):
5669 return self._retry.call_with_retry(
5670 lambda: self._get_connection_and_send_command(*args, **options),
5671 self._reinitialize_on_error,
5672 with_failure_count=True,
5673 )
5675 def _get_connection_and_send_command(self, *args, **options):
5676 redis_node, connection = self._get_client_and_connection_for_transaction()
5678 # Start timing for observability
5679 start_time = time.monotonic()
5681 try:
5682 response = self._send_command_parse_response(
5683 connection, redis_node, args[0], *args, **options
5684 )
5686 record_operation_duration(
5687 command_name=args[0],
5688 duration_seconds=time.monotonic() - start_time,
5689 server_address=connection.host,
5690 server_port=connection.port,
5691 db_namespace=str(connection.db),
5692 )
5694 return response
5695 except Exception as e:
5696 if connection:
5697 # this is used to report the metrics based on host and port info
5698 e.connection = connection
5699 record_operation_duration(
5700 command_name=args[0],
5701 duration_seconds=time.monotonic() - start_time,
5702 server_address=connection.host,
5703 server_port=connection.port,
5704 db_namespace=str(connection.db),
5705 error=e,
5706 )
5707 raise
5709 def _send_command_parse_response(
5710 self, conn, redis_node: Redis, command_name, *args, **options
5711 ):
5712 """
5713 Send a command and parse the response
5714 """
5716 # HIMPORT SET's wire form depends on per-connection state: the fieldset
5717 # must be PREPAREd on this connection first, and any fieldset discarded
5718 # since this connection last reconciled must be dropped. The
5719 # immediate/watched path (commands issued after WATCH, before MULTI)
5720 # would otherwise send a bare HIMPORT SET and fail with "no such
5721 # fieldset". Route it through the node's HIMPORT executor, the same way
5722 # the normal cluster path, the batched MULTI/EXEC path, and standalone
5723 # watched pipelines all do.
5724 himport_set = parse_himport_set_args(args)
5725 if himport_set is not None:
5726 # HIMPORT SET in the joined or split raw form; operands at the right
5727 # offsets. Too few operands returns None and falls through to the bare
5728 # send so the server returns its arity error.
5729 key, fieldset_name, values = himport_set
5730 output = redis_node._himport_execute_set(conn, key, fieldset_name, values)
5731 else:
5732 conn.send_command(*args)
5733 output = redis_node.parse_response(conn, command_name, **options)
5735 if command_name in self.UNWATCH_COMMANDS:
5736 self._watching = False
5737 return output
5739 def _reinitialize_on_error(self, error, failure_count):
5740 if hasattr(error, "connection"):
5741 record_error_count(
5742 server_address=error.connection.host,
5743 server_port=error.connection.port,
5744 network_peer_address=error.connection.host,
5745 network_peer_port=error.connection.port,
5746 error_type=error,
5747 retry_attempts=failure_count,
5748 is_internal=True,
5749 )
5751 if self._watching:
5752 if type(error) in self.SLOT_REDIRECT_ERRORS and self._executing:
5753 raise WatchError("Slot rebalancing occurred while watching keys")
5755 if (
5756 type(error) in self.SLOT_REDIRECT_ERRORS
5757 or type(error) in self.CONNECTION_ERRORS
5758 ):
5759 if self._transaction_connection:
5760 if is_debug_log_enabled():
5761 logger.debug(
5762 f"Operation failed, "
5763 f"with connection: {self._transaction_connection}, "
5764 f"details: {self._transaction_connection.extract_connection_details()}",
5765 )
5766 # Disconnect and release back to pool
5767 self._transaction_connection.disconnect()
5768 node = self._nodes_manager.find_connection_owner(
5769 self._transaction_connection
5770 )
5771 if node and node.redis_connection:
5772 node.redis_connection.connection_pool.release(
5773 self._transaction_connection
5774 )
5775 self._transaction_connection = None
5777 self._pipe.reinitialize_counter += 1
5778 if self._pipe._should_reinitialized():
5779 self._nodes_manager.initialize()
5780 self.reinitialize_counter = 0
5781 else:
5782 if isinstance(error, AskError):
5783 self._nodes_manager.move_slot(error)
5785 self._executing = False
5787 def _raise_first_error(self, responses, stack, start_time):
5788 """
5789 Raise the first exception on the stack
5790 """
5791 for r, cmd in zip(responses, stack):
5792 if isinstance(r, Exception):
5793 self.annotate_exception(r, cmd.position + 1, cmd.args)
5795 record_operation_duration(
5796 command_name="TRANSACTION",
5797 duration_seconds=time.monotonic() - start_time,
5798 server_address=self._transaction_connection.host,
5799 server_port=self._transaction_connection.port,
5800 db_namespace=str(self._transaction_connection.db),
5801 )
5803 raise r
5805 def execute(self, raise_on_error: bool = True) -> List[Any]:
5806 stack = self._command_queue
5807 if not stack and (not self._watching or not self._pipeline_slots):
5808 return []
5810 return self._execute_transaction_with_retries(stack, raise_on_error)
5812 def _execute_transaction_with_retries(
5813 self, stack: List["PipelineCommand"], raise_on_error: bool
5814 ):
5815 return self._retry.call_with_retry(
5816 lambda: self._execute_transaction(stack, raise_on_error),
5817 lambda error, failure_count: self._reinitialize_on_error(
5818 error, failure_count
5819 ),
5820 with_failure_count=True,
5821 )
5823 def _execute_transaction(
5824 self, stack: List["PipelineCommand"], raise_on_error: bool
5825 ):
5826 if len(self._pipeline_slots) > 1:
5827 raise CrossSlotTransactionError(
5828 "All keys involved in a cluster transaction must map to the same slot"
5829 )
5831 self._executing = True
5833 redis_node, connection = self._get_client_and_connection_for_transaction()
5835 # Ensure fieldsets referenced by buffered HIMPORT SETs are prepared on this
5836 # node's connection before the MULTI/EXEC block (session state, not
5837 # transactional). All keys share one slot here, so it is a single node.
5838 self._himport_prepare_pipeline(
5839 redis_node, connection, [(c.args, c.options) for c in stack]
5840 )
5842 stack = chain(
5843 [PipelineCommand(("MULTI",))],
5844 stack,
5845 [PipelineCommand(("EXEC",))],
5846 )
5847 commands = [c.args for c in stack if EMPTY_RESPONSE not in c.options]
5848 packed_commands = connection.pack_commands(commands)
5850 # Start timing for observability
5851 start_time = time.monotonic()
5853 connection.send_packed_command(packed_commands)
5854 errors = []
5856 # parse off the response for MULTI
5857 # NOTE: we need to handle ResponseErrors here and continue
5858 # so that we read all the additional command messages from
5859 # the socket
5860 try:
5861 redis_node.parse_response(connection, "MULTI")
5862 except ResponseError as e:
5863 self.annotate_exception(e, 0, "MULTI")
5864 errors.append(e)
5865 except self.CONNECTION_ERRORS as cluster_error:
5866 self.annotate_exception(cluster_error, 0, "MULTI")
5867 raise
5869 # and all the other commands
5870 for i, command in enumerate(self._command_queue):
5871 if EMPTY_RESPONSE in command.options:
5872 errors.append((i, command.options[EMPTY_RESPONSE]))
5873 else:
5874 try:
5875 _ = redis_node.parse_response(connection, "_")
5876 except self.SLOT_REDIRECT_ERRORS as slot_error:
5877 self.annotate_exception(slot_error, i + 1, command.args)
5878 errors.append(slot_error)
5879 except self.CONNECTION_ERRORS as cluster_error:
5880 self.annotate_exception(cluster_error, i + 1, command.args)
5881 raise
5882 except ResponseError as e:
5883 self.annotate_exception(e, i + 1, command.args)
5884 errors.append(e)
5886 response = None
5887 # parse the EXEC.
5888 try:
5889 response = redis_node.parse_response(connection, "EXEC")
5890 except ExecAbortError:
5891 if errors:
5892 raise errors[0]
5893 raise
5895 self._executing = False
5897 record_operation_duration(
5898 command_name="TRANSACTION",
5899 duration_seconds=time.monotonic() - start_time,
5900 server_address=connection.host,
5901 server_port=connection.port,
5902 db_namespace=str(connection.db),
5903 )
5905 # EXEC clears any watched keys
5906 self._watching = False
5908 if response is None:
5909 raise WatchError("Watched variable changed.")
5911 # put any parse errors into the response
5912 for i, e in errors:
5913 response.insert(i, e)
5915 if len(response) != len(self._command_queue):
5916 raise InvalidPipelineStack(
5917 "Unexpected response length for cluster pipeline EXEC."
5918 " Command stack was {} but response had length {}".format(
5919 [c.args[0] for c in self._command_queue], len(response)
5920 )
5921 )
5923 # find any errors in the response and raise if necessary
5924 if raise_on_error or len(errors) > 0:
5925 self._raise_first_error(
5926 response,
5927 self._command_queue,
5928 start_time,
5929 )
5931 # We have to run response callbacks manually
5932 data = []
5933 for r, cmd in zip(response, self._command_queue):
5934 if not isinstance(r, Exception):
5935 command_name = cmd.args[0]
5936 if command_name in self._pipe.cluster_response_callbacks:
5937 r = self._pipe.cluster_response_callbacks[command_name](
5938 r, **cmd.options
5939 )
5940 data.append(r)
5941 return data
5943 def reset(self):
5944 self._command_queue = []
5946 # make sure to reset the connection state in the event that we were
5947 # watching something
5948 if self._transaction_connection:
5949 try:
5950 if self._watching:
5951 # call this manually since our unwatch or
5952 # immediate_execute_command methods can call reset()
5953 self._transaction_connection.send_command("UNWATCH")
5954 self._transaction_connection.read_response()
5955 # we can safely return the connection to the pool here since we're
5956 # sure we're no longer WATCHing anything
5957 node = self._nodes_manager.find_connection_owner(
5958 self._transaction_connection
5959 )
5960 if node and node.redis_connection:
5961 node.redis_connection.connection_pool.release(
5962 self._transaction_connection
5963 )
5964 self._transaction_connection = None
5965 except self.CONNECTION_ERRORS:
5966 # disconnect will also remove any previous WATCHes
5967 if self._transaction_connection:
5968 self._transaction_connection.disconnect()
5969 node = self._nodes_manager.find_connection_owner(
5970 self._transaction_connection
5971 )
5972 if node and node.redis_connection:
5973 node.redis_connection.connection_pool.release(
5974 self._transaction_connection
5975 )
5976 self._transaction_connection = None
5978 # clean up the other instance attributes
5979 self._watching = False
5980 self._explicit_transaction = False
5981 self._pipeline_slots = set()
5982 self._transaction_has_keyed_slot = False
5983 self._executing = False
5985 def send_cluster_commands(
5986 self, stack, raise_on_error=True, allow_redirections=True
5987 ):
5988 raise NotImplementedError(
5989 "send_cluster_commands cannot be executed in transactional context."
5990 )
5992 def multi(self):
5993 if self._explicit_transaction:
5994 raise RedisError("Cannot issue nested calls to MULTI")
5995 if self._command_queue:
5996 raise RedisError(
5997 "Commands without an initial WATCH have already been issued"
5998 )
5999 self._explicit_transaction = True
6001 def watch(self, *names):
6002 if self._explicit_transaction:
6003 raise RedisError("Cannot issue a WATCH after a MULTI")
6005 return self.execute_command("WATCH", *names)
6007 def unwatch(self):
6008 if self._watching:
6009 return self.execute_command("UNWATCH")
6011 return True
6013 def discard(self):
6014 self.reset()
6016 def delete(self, *names):
6017 return self.execute_command("DEL", *names)
6019 def unlink(self, *names):
6020 return self.execute_command("UNLINK", *names)