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 copy import copy
12from enum import Enum
13from itertools import chain
14from types import MethodType
15from typing import (
16 TYPE_CHECKING,
17 Any,
18 Callable,
19 Dict,
20 Iterable,
21 List,
22 Literal,
23 Optional,
24 Set,
25 Tuple,
26 Type,
27 Union,
28)
30if TYPE_CHECKING:
31 from redis.keyspace_notifications import ClusterKeyspaceNotifications
33from redis import _himport_exec
34from redis._defaults import DEFAULT_RETRY_BASE, DEFAULT_RETRY_CAP, DEFAULT_RETRY_COUNT
35from redis._parsers import CommandsParser, Encoder
36from redis._parsers.commands import CommandPolicies, RequestPolicy, ResponsePolicy
37from redis._parsers.helpers import parse_scan
38from redis.backoff import ExponentialWithJitterBackoff, NoBackoff
39from redis.cache import CacheConfig, CacheFactory, CacheFactoryInterface, CacheInterface
40from redis.client import EMPTY_RESPONSE, CaseInsensitiveDict, PubSub, Redis
41from redis.commands import READ_COMMANDS, RedisClusterCommands
42from redis.commands.helpers import list_or_args, parse_pubsub_subscriptions
43from redis.commands.policies import PolicyResolver, StaticPolicyResolver
44from redis.connection import (
45 Connection,
46 ConnectionPool,
47 parse_url,
48)
49from redis.crc import REDIS_CLUSTER_HASH_SLOTS, key_slot
50from redis.event import (
51 AfterPooledConnectionsInstantiationEvent,
52 AfterPubSubConnectionInstantiationEvent,
53 AfterSlotsCacheRefreshEvent,
54 ClientType,
55 EventDispatcher,
56 EventListenerInterface,
57)
58from redis.exceptions import (
59 AskError,
60 AuthenticationError,
61 ClusterDownError,
62 ClusterError,
63 ConnectionError,
64 CrossSlotTransactionError,
65 DataError,
66 ExecAbortError,
67 InvalidPipelineStack,
68 MaxConnectionsError,
69 MovedError,
70 RedisClusterException,
71 RedisError,
72 ResponseError,
73 SlotNotCoveredError,
74 TimeoutError,
75 TryAgainError,
76 WatchError,
77)
78from redis.himport import HImportRegistry, parse_himport_set_args
79from redis.lock import Lock
80from redis.maint_notifications import (
81 MaintNotificationsConfig,
82 OSSMaintNotificationsHandler,
83)
84from redis.observability.recorder import (
85 record_error_count,
86 record_operation_duration,
87)
88from redis.retry import Retry
89from redis.typing import (
90 ChannelT,
91 FieldT,
92 PubSubHandler,
93 Subscription,
94)
95from redis.utils import (
96 check_protocol_version,
97 deprecated_args,
98 deprecated_function,
99 dict_merge,
100 experimental_method,
101 list_keys_to_dict,
102 merge_result,
103 safe_str,
104 str_if_bytes,
105 truncate_text,
106)
108logger = logging.getLogger(__name__)
111def is_debug_log_enabled():
112 return logger.isEnabledFor(logging.DEBUG)
115def get_node_name(host: str, port: Union[str, int]) -> str:
116 return f"{host}:{port}"
119@deprecated_args(
120 allowed_args=["redis_node"],
121 reason="Use get_connection(redis_node) instead",
122 version="5.3.0",
123)
124def get_connection(redis_node: Redis, *args, **options) -> Connection:
125 return redis_node.connection or redis_node.connection_pool.get_connection()
128def parse_scan_result(command, res, **options):
129 cursors = {}
130 ret = []
131 for node_name, response in res.items():
132 cursor, r = parse_scan(response, **options)
133 cursors[node_name] = cursor
134 ret += r
136 return cursors, ret
139def parse_pubsub_numsub(command, res, **options):
140 numsub_d = OrderedDict()
141 for numsub_tups in res.values():
142 for channel, numsubbed in numsub_tups:
143 try:
144 numsub_d[channel] += numsubbed
145 except KeyError:
146 numsub_d[channel] = numsubbed
148 ret_numsub = [(channel, numsub) for channel, numsub in numsub_d.items()]
149 return ret_numsub
152def parse_cluster_slots(
153 resp: Any, **options: Any
154) -> Dict[Tuple[int, int], Dict[str, Any]]:
155 current_host = options.get("current_host", "")
157 def fix_server(*args: Any) -> Tuple[str, Any]:
158 return str_if_bytes(args[0]) or current_host, args[1]
160 slots = {}
161 for slot in resp:
162 start, end, primary = slot[:3]
163 replicas = slot[3:]
164 slots[start, end] = {
165 "primary": fix_server(*primary),
166 "replicas": [fix_server(*replica) for replica in replicas],
167 }
169 return slots
172def parse_cluster_shards(resp, **options):
173 """
174 Parse CLUSTER SHARDS response.
175 """
176 if isinstance(resp[0], dict):
177 return resp
178 shards = []
179 for x in resp:
180 shard = {"slots": [], "nodes": []}
181 for i in range(0, len(x[1]), 2):
182 shard["slots"].append((x[1][i], (x[1][i + 1])))
183 nodes = x[3]
184 for node in nodes:
185 dict_node = {}
186 for i in range(0, len(node), 2):
187 dict_node[node[i]] = node[i + 1]
188 shard["nodes"].append(dict_node)
189 shards.append(shard)
191 return shards
194def parse_cluster_shards_with_str_keys(resp, **options):
195 """
196 Parse CLUSTER SHARDS with string top-level structural keys.
198 RESP2 parsing exposes top-level shard keys as ``"slots"``/``"nodes"``
199 while node attribute keys keep the connection's decoded/raw form. RESP3 can
200 return top-level shard dictionaries directly, so normalize only the
201 structural shard keys and preserve nested node dictionaries as delivered.
202 """
203 if not resp:
204 return resp
205 if not isinstance(resp[0], dict):
206 return parse_cluster_shards(resp, **options)
208 shards = []
209 for shard_resp in resp:
210 slots = shard_resp.get(b"slots", shard_resp.get("slots", []))
211 nodes = shard_resp.get(b"nodes", shard_resp.get("nodes", []))
212 shard = {
213 "slots": [
214 tuple(slot) if isinstance(slot, list) else slot for slot in slots
215 ],
216 "nodes": [dict(node) if isinstance(node, dict) else node for node in nodes],
217 }
218 shards.append(shard)
219 return shards
222def parse_cluster_shards_unified(resp, **options):
223 """
224 Parse CLUSTER SHARDS into the approved unified shape.
226 Top-level shard keys and nested node attribute keys are strings for both
227 RESP2 and RESP3 wire responses.
228 """
229 if not resp:
230 return resp
231 if isinstance(resp[0], dict):
232 shards = []
233 for shard_resp in resp:
234 slots = shard_resp.get(b"slots", shard_resp.get("slots", []))
235 nodes = shard_resp.get(b"nodes", shard_resp.get("nodes", []))
236 shard = {
237 "slots": slots,
238 "nodes": [
239 {str_if_bytes(k): v for k, v in node.items()}
240 if isinstance(node, dict)
241 else node
242 for node in nodes
243 ],
244 }
245 shards.append(shard)
246 return shards
248 shards = []
249 for x in resp:
250 shard = {"slots": [], "nodes": []}
251 for i in range(0, len(x[1]), 2):
252 shard["slots"].append((x[1][i], x[1][i + 1]))
253 nodes = x[3]
254 for node in nodes:
255 dict_node = {}
256 for i in range(0, len(node), 2):
257 dict_node[str_if_bytes(node[i])] = node[i + 1]
258 shard["nodes"].append(dict_node)
259 shards.append(shard)
260 return shards
263def parse_cluster_myshardid(resp, **options):
264 """
265 Parse CLUSTER MYSHARDID response.
266 """
267 return resp.decode("utf-8")
270PRIMARY = "primary"
271REPLICA = "replica"
272SLOT_ID = "slot-id"
274REDIS_ALLOWED_KEYS = (
275 "connection_class",
276 "connection_pool",
277 "connection_pool_class",
278 "client_name",
279 "credential_provider",
280 "db",
281 "decode_responses",
282 "encoding",
283 "encoding_errors",
284 "host",
285 "driver_info",
286 "lib_name",
287 "lib_version",
288 "max_connections",
289 "nodes_flag",
290 "redis_connect_func",
291 "password",
292 "port",
293 "timeout",
294 "queue_class",
295 "retry",
296 "retry_on_timeout",
297 "protocol",
298 "legacy_responses",
299 "socket_connect_timeout",
300 "socket_keepalive",
301 "socket_keepalive_options",
302 "socket_read_size",
303 "socket_timeout",
304 "ssl",
305 "ssl_ca_certs",
306 "ssl_ca_data",
307 "ssl_ca_path",
308 "ssl_certfile",
309 "ssl_cert_reqs",
310 "ssl_include_verify_flags",
311 "ssl_exclude_verify_flags",
312 "ssl_keyfile",
313 "ssl_password",
314 "ssl_check_hostname",
315 "unix_socket_path",
316 "username",
317 "cache",
318 "cache_config",
319 "maint_notifications_config",
320)
321KWARGS_DISABLED_KEYS = ("host", "port", "retry")
324def cleanup_kwargs(**kwargs):
325 """
326 Remove unsupported or disabled keys from kwargs
327 """
328 connection_kwargs = {
329 k: v
330 for k, v in kwargs.items()
331 if k in REDIS_ALLOWED_KEYS and k not in KWARGS_DISABLED_KEYS
332 }
334 return connection_kwargs
337class MaintNotificationsAbstractRedisCluster:
338 """
339 Abstract class for handling maintenance notifications logic.
340 This class is expected to be used as base class together with RedisCluster.
342 This class is intended to be used with multiple inheritance!
344 All logic related to maintenance notifications is encapsulated in this class.
345 """
347 def __init__(
348 self,
349 maint_notifications_config: Optional[MaintNotificationsConfig],
350 **kwargs,
351 ):
352 # Initialize maintenance notifications.
353 # The RESP3 requirement is validated in RedisCluster.__init__ before the
354 # NodesManager is constructed; this mixin is only ever run from there, so
355 # the config it receives has already been validated.
356 is_protocol_supported = check_protocol_version(kwargs.get("protocol"), 3)
358 if maint_notifications_config is None and is_protocol_supported:
359 maint_notifications_config = MaintNotificationsConfig()
361 self.maint_notifications_config = maint_notifications_config
363 if self.maint_notifications_config and self.maint_notifications_config.enabled:
364 self._oss_cluster_maint_notifications_handler = (
365 OSSMaintNotificationsHandler(self, self.maint_notifications_config)
366 )
367 # Update connection kwargs for all future nodes connections
368 self._update_connection_kwargs_for_maint_notifications(
369 self._oss_cluster_maint_notifications_handler
370 )
371 # Update existing nodes connections - they are created as part of the RedisCluster constructor
372 for node in self.get_nodes():
373 if node.redis_connection is None:
374 continue
375 node.redis_connection.connection_pool.update_maint_notifications_config(
376 self.maint_notifications_config,
377 oss_cluster_maint_notifications_handler=self._oss_cluster_maint_notifications_handler,
378 )
379 else:
380 self._oss_cluster_maint_notifications_handler = None
382 def _update_connection_kwargs_for_maint_notifications(
383 self, oss_cluster_maint_notifications_handler: OSSMaintNotificationsHandler
384 ):
385 """
386 Update the connection kwargs for all future connections.
387 """
388 self.nodes_manager.connection_kwargs.update(
389 {
390 "oss_cluster_maint_notifications_handler": oss_cluster_maint_notifications_handler,
391 }
392 )
395class AbstractRedisCluster:
396 RedisClusterRequestTTL = 16
398 PRIMARIES = "primaries"
399 REPLICAS = "replicas"
400 ALL_NODES = "all"
401 RANDOM = "random"
402 DEFAULT_NODE = "default-node"
404 NODE_FLAGS = {PRIMARIES, REPLICAS, ALL_NODES, RANDOM, DEFAULT_NODE}
406 COMMAND_FLAGS = dict_merge(
407 list_keys_to_dict(
408 [
409 "ACL CAT",
410 "ACL DELUSER",
411 "ACL DRYRUN",
412 "ACL GENPASS",
413 "ACL GETUSER",
414 "ACL HELP",
415 "ACL LIST",
416 "ACL LOG",
417 "ACL LOAD",
418 "ACL SAVE",
419 "ACL SETUSER",
420 "ACL USERS",
421 "ACL WHOAMI",
422 "AUTH",
423 "CLIENT LIST",
424 "CLIENT SETINFO",
425 "CLIENT SETNAME",
426 "CLIENT GETNAME",
427 "CONFIG SET",
428 "CONFIG REWRITE",
429 "CONFIG RESETSTAT",
430 "TIME",
431 "PUBSUB CHANNELS",
432 "PUBSUB NUMPAT",
433 "PUBSUB NUMSUB",
434 "PUBSUB SHARDCHANNELS",
435 "PUBSUB SHARDNUMSUB",
436 "PING",
437 "INFO",
438 "SHUTDOWN",
439 "KEYS",
440 "DBSIZE",
441 "BGSAVE",
442 "SLOWLOG GET",
443 "SLOWLOG LEN",
444 "SLOWLOG RESET",
445 "WAIT",
446 "WAITAOF",
447 "SAVE",
448 "MEMORY PURGE",
449 "MEMORY MALLOC-STATS",
450 "MEMORY STATS",
451 "LASTSAVE",
452 "CLIENT TRACKINGINFO",
453 "CLIENT PAUSE",
454 "CLIENT UNPAUSE",
455 "CLIENT UNBLOCK",
456 "CLIENT ID",
457 "CLIENT REPLY",
458 "CLIENT GETREDIR",
459 "CLIENT INFO",
460 "CLIENT KILL",
461 "READONLY",
462 "CLUSTER INFO",
463 "CLUSTER MEET",
464 "CLUSTER MYSHARDID",
465 "CLUSTER NODES",
466 "CLUSTER REPLICAS",
467 "CLUSTER RESET",
468 "CLUSTER SET-CONFIG-EPOCH",
469 "CLUSTER SLOTS",
470 "CLUSTER SHARDS",
471 "CLUSTER COUNT-FAILURE-REPORTS",
472 "CLUSTER KEYSLOT",
473 "COMMAND",
474 "COMMAND COUNT",
475 "COMMAND LIST",
476 "COMMAND GETKEYS",
477 "CONFIG GET",
478 "DEBUG",
479 "RANDOMKEY",
480 "READONLY",
481 "READWRITE",
482 "TIME",
483 "TFUNCTION LOAD",
484 "TFUNCTION DELETE",
485 "TFUNCTION LIST",
486 "TFCALL",
487 "TFCALLASYNC",
488 "LATENCY HISTORY",
489 "LATENCY LATEST",
490 "LATENCY RESET",
491 "MODULE LIST",
492 "MODULE LOAD",
493 "MODULE UNLOAD",
494 "MODULE LOADEX",
495 ],
496 DEFAULT_NODE,
497 ),
498 list_keys_to_dict(
499 [
500 "FLUSHALL",
501 "FLUSHDB",
502 "FUNCTION DELETE",
503 "FUNCTION FLUSH",
504 "FUNCTION LIST",
505 "FUNCTION LOAD",
506 "FUNCTION RESTORE",
507 "SCAN",
508 "SCRIPT EXISTS",
509 "SCRIPT FLUSH",
510 "SCRIPT LOAD",
511 ],
512 PRIMARIES,
513 ),
514 list_keys_to_dict(["FUNCTION DUMP"], RANDOM),
515 list_keys_to_dict(
516 [
517 "CLUSTER COUNTKEYSINSLOT",
518 "CLUSTER DELSLOTS",
519 "CLUSTER DELSLOTSRANGE",
520 "CLUSTER GETKEYSINSLOT",
521 "CLUSTER SETSLOT",
522 ],
523 SLOT_ID,
524 ),
525 )
527 SEARCH_COMMANDS = (
528 [
529 "FT.CREATE",
530 "FT.SEARCH",
531 "FT.AGGREGATE",
532 "FT.EXPLAIN",
533 "FT.EXPLAINCLI",
534 "FT,PROFILE",
535 "FT.ALTER",
536 "FT.DROPINDEX",
537 "FT.ALIASADD",
538 "FT.ALIASUPDATE",
539 "FT.ALIASDEL",
540 "FT.ALIASLIST",
541 "FT.TAGVALS",
542 "FT.SUGADD",
543 "FT.SUGGET",
544 "FT.SUGDEL",
545 "FT.SUGLEN",
546 "FT.SYNUPDATE",
547 "FT.SYNDUMP",
548 "FT.SPELLCHECK",
549 "FT.DICTADD",
550 "FT.DICTDEL",
551 "FT.DICTDUMP",
552 "FT.INFO",
553 "FT._LIST",
554 "FT.CONFIG",
555 "FT.ADD",
556 "FT.DEL",
557 "FT.DROP",
558 "FT.GET",
559 "FT.MGET",
560 "FT.SYNADD",
561 ],
562 )
564 CLUSTER_COMMANDS_RESPONSE_CALLBACKS = {
565 "CLUSTER SLOTS": parse_cluster_slots,
566 "CLUSTER SHARDS": parse_cluster_shards,
567 "CLUSTER MYSHARDID": parse_cluster_myshardid,
568 }
570 RESULT_CALLBACKS = dict_merge(
571 list_keys_to_dict(["PUBSUB NUMSUB", "PUBSUB SHARDNUMSUB"], parse_pubsub_numsub),
572 list_keys_to_dict(
573 ["PUBSUB NUMPAT"], lambda command, res: sum(list(res.values()))
574 ),
575 list_keys_to_dict(
576 ["KEYS", "PUBSUB CHANNELS", "PUBSUB SHARDCHANNELS"], merge_result
577 ),
578 list_keys_to_dict(
579 [
580 "PING",
581 "CONFIG SET",
582 "CONFIG REWRITE",
583 "CONFIG RESETSTAT",
584 "CLIENT SETNAME",
585 "BGSAVE",
586 "SLOWLOG RESET",
587 "SAVE",
588 "MEMORY PURGE",
589 "CLIENT PAUSE",
590 "CLIENT UNPAUSE",
591 ],
592 lambda command, res: all(res.values()) if isinstance(res, dict) else res,
593 ),
594 list_keys_to_dict(
595 ["DBSIZE", "WAIT"],
596 lambda command, res: sum(res.values()) if isinstance(res, dict) else res,
597 ),
598 list_keys_to_dict(
599 ["CLIENT UNBLOCK"], lambda command, res: 1 if sum(res.values()) > 0 else 0
600 ),
601 list_keys_to_dict(["SCAN"], parse_scan_result),
602 list_keys_to_dict(
603 ["SCRIPT LOAD"], lambda command, res: list(res.values()).pop()
604 ),
605 list_keys_to_dict(
606 ["SCRIPT EXISTS"], lambda command, res: [all(k) for k in zip(*res.values())]
607 ),
608 list_keys_to_dict(["SCRIPT FLUSH"], lambda command, res: all(res.values())),
609 )
611 ERRORS_ALLOW_RETRY = (
612 ConnectionError,
613 TimeoutError,
614 ClusterDownError,
615 SlotNotCoveredError,
616 )
618 def replace_default_node(self, target_node: "ClusterNode" = None) -> None:
619 """Replace the default cluster node.
620 A random cluster node will be chosen if target_node isn't passed, and primaries
621 will be prioritized. The default node will not be changed if there are no other
622 nodes in the cluster.
624 Args:
625 target_node (ClusterNode, optional): Target node to replace the default
626 node. Defaults to None.
627 """
628 if target_node:
629 self.nodes_manager.default_node = target_node
630 else:
631 curr_node = self.get_default_node()
632 primaries = [node for node in self.get_primaries() if node != curr_node]
633 if primaries:
634 # Choose a primary if the cluster contains different primaries
635 self.nodes_manager.default_node = random.choice(primaries)
636 else:
637 # Otherwise, choose a primary if the cluster contains different primaries
638 replicas = [node for node in self.get_replicas() if node != curr_node]
639 if replicas:
640 self.nodes_manager.default_node = random.choice(replicas)
643class RedisCluster(
644 AbstractRedisCluster, MaintNotificationsAbstractRedisCluster, RedisClusterCommands
645):
646 # Type discrimination marker for @overload self-type pattern
647 _is_async_client: Literal[False] = False
649 @classmethod
650 def from_url(cls, url: str, **kwargs: Any) -> "RedisCluster":
651 """
652 Return a Redis client object configured from the given URL
654 For example::
656 redis://[[username]:[password]]@localhost:6379/0
657 rediss://[[username]:[password]]@localhost:6379/0
658 unix://[username@]/path/to/socket.sock?db=0[&password=password]
660 Three URL schemes are supported:
662 - `redis://` creates a TCP socket connection. See more at:
663 <https://www.iana.org/assignments/uri-schemes/prov/redis>
664 - `rediss://` creates a SSL wrapped TCP socket connection. See more at:
665 <https://www.iana.org/assignments/uri-schemes/prov/rediss>
666 - ``unix://``: creates a Unix Domain Socket connection.
668 The username, password, hostname and path are passed through
669 urllib.parse.unquote in order to replace any percent-encoded values
670 with their corresponding characters. Querystring values are decoded
671 by urllib.parse.parse_qs and are not unquoted again.
673 There are several ways to specify a database number. The first value
674 found will be used:
676 1. A ``db`` querystring option, e.g. redis://localhost?db=0
677 2. If using the redis:// or rediss:// schemes, the path argument
678 of the url, e.g. redis://localhost/0
679 3. A ``db`` keyword argument to this function.
681 If none of these options are specified, the default db=0 is used.
683 All querystring options are cast to their appropriate Python types.
684 Boolean arguments can be specified with string values "True"/"False"
685 or "Yes"/"No". Values that cannot be properly cast cause a
686 ``ValueError`` to be raised. Once parsed, the querystring arguments
687 and keyword arguments are passed to the ``ConnectionPool``'s
688 class initializer. In the case of conflicting arguments, querystring
689 arguments always win.
691 """
692 return cls(url=url, **kwargs)
694 @deprecated_args(
695 args_to_warn=["read_from_replicas"],
696 reason="Please configure the 'load_balancing_strategy' instead",
697 version="5.3.0",
698 )
699 @deprecated_args(
700 args_to_warn=[
701 "cluster_error_retry_attempts",
702 ],
703 reason="Please configure the 'retry' object instead",
704 version="6.0.0",
705 )
706 def __init__(
707 self,
708 host: Optional[str] = None,
709 port: int = 6379,
710 startup_nodes: Optional[List["ClusterNode"]] = None,
711 cluster_error_retry_attempts: int = DEFAULT_RETRY_COUNT,
712 retry: Optional["Retry"] = None,
713 require_full_coverage: bool = True,
714 reinitialize_steps: int = 5,
715 read_from_replicas: bool = False,
716 load_balancing_strategy: Optional["LoadBalancingStrategy"] = None,
717 dynamic_startup_nodes: bool = True,
718 url: Optional[str] = None,
719 address_remap: Optional[Callable[[Tuple[str, int]], Tuple[str, int]]] = None,
720 cache: Optional[CacheInterface] = None,
721 cache_config: Optional[CacheConfig] = None,
722 event_dispatcher: Optional[EventDispatcher] = None,
723 policy_resolver: PolicyResolver = StaticPolicyResolver(),
724 maint_notifications_config: Optional[MaintNotificationsConfig] = None,
725 **kwargs,
726 ):
727 """
728 Initialize a new RedisCluster client.
730 :param startup_nodes:
731 List of nodes from which initial bootstrapping can be done
732 :param host:
733 Can be used to point to a startup node
734 :param port:
735 Can be used to point to a startup node
736 :param require_full_coverage:
737 When set to False (default value): the client will not require a
738 full coverage of the slots. However, if not all slots are covered,
739 and at least one node has 'cluster-require-full-coverage' set to
740 'yes,' the server will throw a ClusterDownError for some key-based
741 commands. See -
742 https://redis.io/topics/cluster-tutorial#redis-cluster-configuration-parameters
743 When set to True: all slots must be covered to construct the
744 cluster client. If not all slots are covered, RedisClusterException
745 will be thrown.
746 :param read_from_replicas:
747 @deprecated - please use load_balancing_strategy instead
748 Enable read from replicas in READONLY mode. You can read possibly
749 stale data.
750 When set to true, read commands will be assigned between the
751 primary and its replications in a Round-Robin manner.
752 :param load_balancing_strategy:
753 Enable read from replicas in READONLY mode and defines the load balancing
754 strategy that will be used for cluster node selection.
755 The data read from replicas is eventually consistent with the data in primary nodes.
756 :param dynamic_startup_nodes:
757 Set the RedisCluster's startup nodes to all of the discovered nodes.
758 If true (default value), the cluster's discovered nodes will be used to
759 determine the cluster nodes-slots mapping in the next topology refresh.
760 It will remove the initial passed startup nodes if their endpoints aren't
761 listed in the CLUSTER SLOTS output.
762 If you use dynamic DNS endpoints for startup nodes but CLUSTER SLOTS lists
763 specific IP addresses, it is best to set it to false.
764 :param cluster_error_retry_attempts:
765 @deprecated - Please configure the 'retry' object instead
766 In case 'retry' object is set - this argument is ignored!
768 Number of times to retry before raising an error when
769 :class:`~.TimeoutError` or :class:`~.ConnectionError`, :class:`~.SlotNotCoveredError` or
770 :class:`~.ClusterDownError` are encountered
771 :param retry:
772 A retry object that defines the retry strategy and the number of
773 retries for the cluster client.
774 In current implementation for the cluster client (starting form redis-py version 6.0.0)
775 the retry object is not yet fully utilized, instead it is used just to determine
776 the number of retries for the cluster client.
777 In the future releases the retry object will be used to handle the cluster client retries!
778 :param reinitialize_steps:
779 Specifies the number of MOVED errors that need to occur before
780 reinitializing the whole cluster topology. If a MOVED error occurs
781 and the cluster does not need to be reinitialized on this current
782 error handling, only the MOVED slot will be patched with the
783 redirected node.
784 To reinitialize the cluster on every MOVED error, set
785 reinitialize_steps to 1.
786 To avoid reinitializing the cluster on moved errors, set
787 reinitialize_steps to 0.
788 :param address_remap:
789 An optional callable which, when provided with an internal network
790 address of a node, e.g. a `(host, port)` tuple, will return the address
791 where the node is reachable. This can be used to map the addresses at
792 which the nodes _think_ they are, to addresses at which a client may
793 reach them, such as when they sit behind a proxy.
795 :param maint_notifications_config:
796 Configures the nodes connections to support maintenance notifications - see
797 `redis.maint_notifications.MaintNotificationsConfig` for details.
798 Only supported with RESP3.
799 If not provided and protocol is RESP3, the maintenance notifications
800 will be enabled by default (logic is included in the NodesManager
801 initialization).
802 :**kwargs:
803 Extra arguments that will be sent into Redis instance when created
804 (See Official redis-py doc for supported kwargs - the only limitation
805 is that you can't provide 'retry' object as part of kwargs.
806 [https://github.com/andymccurdy/redis-py/blob/master/redis/client.py])
807 Some kwargs are not supported and will raise a
808 RedisClusterException:
809 - db (Redis do not support database SELECT in cluster mode)
811 """
812 if startup_nodes is None:
813 startup_nodes = []
815 if "db" in kwargs:
816 # Argument 'db' is not possible to use in cluster mode
817 raise RedisClusterException(
818 "Argument 'db' is not possible to use in cluster mode"
819 )
821 if "retry" in kwargs:
822 # Argument 'retry' is not possible to be used in kwargs when in cluster mode
823 # the kwargs are set to the lower level connections to the cluster nodes
824 # and there we provide retry configuration without retries allowed.
825 # The retries should be handled on cluster client level.
826 raise RedisClusterException(
827 "The 'retry' argument cannot be used in kwargs when running in cluster mode."
828 )
830 # Get the startup node/s
831 from_url = False
832 if url is not None:
833 from_url = True
834 url_options = parse_url(url)
835 if "path" in url_options:
836 raise RedisClusterException(
837 "RedisCluster does not currently support Unix Domain "
838 "Socket connections"
839 )
840 if "db" in url_options and url_options["db"] != 0:
841 # Argument 'db' is not possible to use in cluster mode
842 raise RedisClusterException(
843 "A ``db`` querystring option can only be 0 in cluster mode"
844 )
845 kwargs.update(url_options)
846 host = kwargs.get("host")
847 port = kwargs.get("port", port)
848 startup_nodes.append(ClusterNode(host, port))
849 elif host is not None and port is not None:
850 startup_nodes.append(ClusterNode(host, port))
851 elif len(startup_nodes) == 0:
852 # No startup node was provided
853 raise RedisClusterException(
854 "RedisCluster requires at least one node to discover the "
855 "cluster. Please provide one of the followings:\n"
856 "1. host and port, for example:\n"
857 " RedisCluster(host='localhost', port=6379)\n"
858 "2. list of startup nodes, for example:\n"
859 " RedisCluster(startup_nodes=[ClusterNode('localhost', 6379),"
860 " ClusterNode('localhost', 6378)])"
861 )
862 # Update the connection arguments
863 # Whenever a new connection is established, RedisCluster's on_connect
864 # method should be run
865 # If the user passed on_connect function we'll save it and run it
866 # inside the RedisCluster.on_connect() function
867 self.user_on_connect_func = kwargs.pop("redis_connect_func", None)
868 kwargs.update({"redis_connect_func": self.on_connect})
869 kwargs = cleanup_kwargs(**kwargs)
870 if retry:
871 self.retry = retry
872 else:
873 self.retry = Retry(
874 backoff=ExponentialWithJitterBackoff(
875 base=DEFAULT_RETRY_BASE, cap=DEFAULT_RETRY_CAP
876 ),
877 retries=cluster_error_retry_attempts,
878 )
880 self.encoder = Encoder(
881 kwargs.get("encoding", "utf-8"),
882 kwargs.get("encoding_errors", "strict"),
883 kwargs.get("decode_responses", False),
884 )
885 protocol = kwargs.get("protocol", None)
886 if (cache_config or cache) and not check_protocol_version(protocol, 3):
887 raise RedisError("Client caching is only supported with RESP version 3")
889 if (
890 maint_notifications_config
891 and maint_notifications_config.enabled
892 and not check_protocol_version(protocol, 3)
893 ):
894 raise RedisError(
895 "Maintenance notifications are only supported with RESP version 3"
896 )
897 if check_protocol_version(protocol, 3) and maint_notifications_config is None:
898 maint_notifications_config = MaintNotificationsConfig()
900 # Build the client-level HIMPORT registry once (always empty at construction)
901 # and share the same object with every node pool, so the fieldset registry is
902 # shared cluster-wide and runtime himport_prepare mutates one object. It is
903 # handed to the NodesManager and injected onto each node's pool in
904 # create_redis_node; it is deliberately NOT forwarded through connection_kwargs,
905 # so nodes reuse the one shared object rather than each rebuilding their own.
906 self._himport_registry = HImportRegistry()
908 self.command_flags = self.__class__.COMMAND_FLAGS.copy()
909 self.node_flags = self.__class__.NODE_FLAGS.copy()
910 self.read_from_replicas = read_from_replicas
911 self.load_balancing_strategy = load_balancing_strategy
912 self.reinitialize_counter = 0
913 self.reinitialize_steps = reinitialize_steps
914 if event_dispatcher is None:
915 self._event_dispatcher = EventDispatcher()
916 else:
917 self._event_dispatcher = event_dispatcher
918 self.startup_nodes = startup_nodes
920 self.nodes_manager = NodesManager(
921 startup_nodes=startup_nodes,
922 from_url=from_url,
923 require_full_coverage=require_full_coverage,
924 dynamic_startup_nodes=dynamic_startup_nodes,
925 address_remap=address_remap,
926 cache=cache,
927 cache_config=cache_config,
928 event_dispatcher=self._event_dispatcher,
929 maint_notifications_config=maint_notifications_config,
930 himport_registry=self._himport_registry,
931 **kwargs,
932 )
934 cluster_response_callbacks = dict(
935 self.__class__.CLUSTER_COMMANDS_RESPONSE_CALLBACKS
936 )
937 legacy_responses = kwargs.get("legacy_responses", True)
938 protocol = kwargs.get("protocol")
939 if not legacy_responses:
940 cluster_response_callbacks["CLUSTER SHARDS"] = parse_cluster_shards_unified
941 elif protocol is None:
942 cluster_response_callbacks["CLUSTER SHARDS"] = (
943 parse_cluster_shards_with_str_keys
944 )
945 self.cluster_response_callbacks = CaseInsensitiveDict(
946 cluster_response_callbacks
947 )
948 self.result_callbacks = CaseInsensitiveDict(self.__class__.RESULT_CALLBACKS)
950 # For backward compatibility, mapping from existing policies to new one
951 self._command_flags_mapping: dict[str, Union[RequestPolicy, ResponsePolicy]] = {
952 self.__class__.RANDOM: RequestPolicy.DEFAULT_KEYLESS,
953 self.__class__.PRIMARIES: RequestPolicy.ALL_SHARDS,
954 self.__class__.ALL_NODES: RequestPolicy.ALL_NODES,
955 self.__class__.REPLICAS: RequestPolicy.ALL_REPLICAS,
956 self.__class__.DEFAULT_NODE: RequestPolicy.DEFAULT_NODE,
957 SLOT_ID: RequestPolicy.DEFAULT_KEYED,
958 }
960 self._policies_callback_mapping: dict[
961 Union[RequestPolicy, ResponsePolicy], Callable
962 ] = {
963 RequestPolicy.DEFAULT_KEYLESS: lambda command_name: [
964 self.get_random_primary_or_all_nodes(command_name)
965 ],
966 RequestPolicy.DEFAULT_KEYED: lambda command,
967 *args: self.get_nodes_from_slot(command, *args),
968 RequestPolicy.DEFAULT_NODE: lambda: [self.get_default_node()],
969 RequestPolicy.ALL_SHARDS: self.get_primaries,
970 RequestPolicy.ALL_NODES: self.get_nodes,
971 RequestPolicy.ALL_REPLICAS: self.get_replicas,
972 RequestPolicy.MULTI_SHARD: lambda *args,
973 **kwargs: self._split_multi_shard_command(*args, **kwargs),
974 RequestPolicy.SPECIAL: self.get_special_nodes,
975 ResponsePolicy.DEFAULT_KEYLESS: lambda res: res,
976 ResponsePolicy.DEFAULT_KEYED: lambda res: res,
977 }
979 self._policy_resolver = policy_resolver
980 self.commands_parser = CommandsParser(self)
982 # Node where FT.AGGREGATE command is executed.
983 self._aggregate_nodes = None
984 self._lock = threading.RLock()
986 MaintNotificationsAbstractRedisCluster.__init__(
987 self, maint_notifications_config, **kwargs
988 )
990 def __enter__(self):
991 return self
993 def __exit__(self, exc_type, exc_value, traceback):
994 self.close()
996 def __del__(self):
997 try:
998 self.close()
999 except Exception:
1000 pass
1002 def disconnect_connection_pools(self):
1003 for node in self.get_nodes():
1004 if node.redis_connection:
1005 try:
1006 node.redis_connection.connection_pool.disconnect()
1007 except OSError:
1008 # Client was already disconnected. do nothing
1009 pass
1011 def on_connect(self, connection):
1012 """
1013 Initialize the connection, authenticate and select a database and send
1014 READONLY if it is set during object initialization.
1015 """
1016 connection.on_connect()
1018 if self.read_from_replicas or self.load_balancing_strategy:
1019 # Sending READONLY command to server to configure connection as
1020 # readonly. Since each cluster node may change its server type due
1021 # to a failover, we should establish a READONLY connection
1022 # regardless of the server type. If this is a primary connection,
1023 # READONLY would not affect executing write commands.
1024 connection.send_command("READONLY")
1025 if str_if_bytes(connection.read_response()) != "OK":
1026 raise ConnectionError("READONLY command failed")
1028 if self.user_on_connect_func is not None:
1029 self.user_on_connect_func(connection)
1031 def get_redis_connection(self, node: "ClusterNode") -> Redis:
1032 if not node.redis_connection:
1033 with self._lock:
1034 if not node.redis_connection:
1035 self.nodes_manager.create_redis_connections([node])
1036 return node.redis_connection
1038 def get_node(self, host=None, port=None, node_name=None):
1039 return self.nodes_manager.get_node(host, port, node_name)
1041 def get_primaries(self):
1042 return self.nodes_manager.get_nodes_by_server_type(PRIMARY)
1044 def get_replicas(self):
1045 return self.nodes_manager.get_nodes_by_server_type(REPLICA)
1047 def get_random_node(self):
1048 return random.choice(list(self.nodes_manager.nodes_cache.values()))
1050 def get_random_primary_or_all_nodes(self, command_name):
1051 """
1052 Returns random primary or all nodes depends on READONLY mode.
1053 """
1054 if self.read_from_replicas and command_name in READ_COMMANDS:
1055 return self.get_random_node()
1057 return self.get_random_primary_node()
1059 def get_nodes(self):
1060 return list(self.nodes_manager.nodes_cache.values())
1062 def get_node_from_key(self, key, replica=False):
1063 """
1064 Get the node that holds the key's slot.
1065 If replica set to True but the slot doesn't have any replicas, None is
1066 returned.
1067 """
1068 slot = self.keyslot(key)
1069 slot_cache = self.nodes_manager.slots_cache.get(slot)
1070 if slot_cache is None or len(slot_cache) == 0:
1071 raise SlotNotCoveredError(f'Slot "{slot}" is not covered by the cluster.')
1072 if replica and len(self.nodes_manager.slots_cache[slot]) < 2:
1073 return None
1074 elif replica:
1075 node_idx = 1
1076 else:
1077 # primary
1078 node_idx = 0
1080 return slot_cache[node_idx]
1082 def get_default_node(self):
1083 """
1084 Get the cluster's default node
1085 """
1086 return self.nodes_manager.default_node
1088 def get_nodes_from_slot(self, command: str, *args):
1089 """
1090 Returns a list of nodes that hold the specified keys' slots.
1091 """
1092 # get the node that holds the key's slot
1093 slot = self.determine_slot(*args)
1094 node = self.nodes_manager.get_node_from_slot(
1095 slot,
1096 self.read_from_replicas and command in READ_COMMANDS,
1097 self.load_balancing_strategy if command in READ_COMMANDS else None,
1098 )
1099 return [node]
1101 def _split_multi_shard_command(self, *args, **kwargs) -> list[dict]:
1102 """
1103 Splits the command with Multi-Shard policy, to the multiple commands
1104 """
1105 keys = self._get_command_keys(*args)
1106 commands = []
1108 for key in keys:
1109 commands.append(
1110 {
1111 "args": (args[0], key),
1112 "kwargs": kwargs,
1113 }
1114 )
1116 return commands
1118 def get_special_nodes(self) -> Optional[list["ClusterNode"]]:
1119 """
1120 Returns a list of nodes for commands with a special policy.
1121 """
1122 if not self._aggregate_nodes:
1123 raise RedisClusterException(
1124 "Cannot execute FT.CURSOR commands without FT.AGGREGATE"
1125 )
1127 return self._aggregate_nodes
1129 def get_random_primary_node(self) -> "ClusterNode":
1130 """
1131 Returns a random primary node
1132 """
1133 return random.choice(self.get_primaries())
1135 def _evaluate_all_succeeded(self, res):
1136 """
1137 Evaluate the result of a command with ResponsePolicy.ALL_SUCCEEDED
1138 """
1139 first_successful_response = None
1141 if isinstance(res, dict):
1142 for key, value in res.items():
1143 if value:
1144 if first_successful_response is None:
1145 first_successful_response = {key: value}
1146 else:
1147 return {key: False}
1148 else:
1149 for response in res:
1150 if response:
1151 if first_successful_response is None:
1152 # Dynamically resolve type
1153 first_successful_response = type(response)(response)
1154 else:
1155 return type(response)(False)
1157 return first_successful_response
1159 def set_default_node(self, node):
1160 """
1161 Set the default node of the cluster.
1162 :param node: 'ClusterNode'
1163 :return True if the default node was set, else False
1164 """
1165 if node is None or self.get_node(node_name=node.name) is None:
1166 return False
1167 self.nodes_manager.default_node = node
1168 return True
1170 def set_retry(self, retry: Retry) -> None:
1171 self.retry = retry
1173 def monitor(self, target_node=None):
1174 """
1175 Returns a Monitor object for the specified target node.
1176 The default cluster node will be selected if no target node was
1177 specified.
1178 Monitor is useful for handling the MONITOR command to the redis server.
1179 next_command() method returns one command from monitor
1180 listen() method yields commands from monitor.
1181 """
1182 if target_node is None:
1183 target_node = self.get_default_node()
1184 if target_node.redis_connection is None:
1185 raise RedisClusterException(
1186 f"Cluster Node {target_node.name} has no redis_connection"
1187 )
1188 return target_node.redis_connection.monitor()
1190 def pubsub(self, node=None, host=None, port=None, **kwargs):
1191 """
1192 Allows passing a ClusterNode, or host&port, to get a pubsub instance
1193 connected to the specified node
1194 """
1195 return ClusterPubSub(self, node=node, host=host, port=port, **kwargs)
1197 def keyspace_notifications(
1198 self,
1199 key_prefix: Union[str, bytes, None] = None,
1200 ignore_subscribe_messages: bool = True,
1201 ) -> "ClusterKeyspaceNotifications":
1202 """
1203 Return a :class:`~redis.keyspace_notifications.ClusterKeyspaceNotifications`
1204 object for subscribing to keyspace and keyevent notifications across
1205 all primary nodes in the cluster.
1207 Note: Keyspace notifications must be enabled on all Redis cluster nodes
1208 via the ``notify-keyspace-events`` configuration option.
1210 Args:
1211 key_prefix: Optional prefix to filter and strip from keys in
1212 notifications.
1213 ignore_subscribe_messages: If True, subscribe/unsubscribe
1214 confirmations are not returned by
1215 get_message/listen.
1216 """
1217 from redis.keyspace_notifications import ClusterKeyspaceNotifications
1219 return ClusterKeyspaceNotifications(
1220 self,
1221 key_prefix=key_prefix,
1222 ignore_subscribe_messages=ignore_subscribe_messages,
1223 )
1225 def pipeline(self, transaction=None, shard_hint=None):
1226 """
1227 Cluster impl:
1228 Pipelines do not work in cluster mode the same way they
1229 do in normal mode. Create a clone of this object so
1230 that simulating pipelines will work correctly. Each
1231 command will be called directly when used and
1232 when calling execute() will only return the result stack.
1233 """
1234 if shard_hint:
1235 raise RedisClusterException("shard_hint is deprecated in cluster mode")
1237 return ClusterPipeline(
1238 nodes_manager=self.nodes_manager,
1239 commands_parser=self.commands_parser,
1240 startup_nodes=self.nodes_manager.startup_nodes,
1241 result_callbacks=self.result_callbacks,
1242 cluster_response_callbacks=self.cluster_response_callbacks,
1243 read_from_replicas=self.read_from_replicas,
1244 load_balancing_strategy=self.load_balancing_strategy,
1245 reinitialize_steps=self.reinitialize_steps,
1246 retry=self.retry,
1247 lock=self._lock,
1248 transaction=transaction,
1249 event_dispatcher=self._event_dispatcher,
1250 )
1252 def lock(
1253 self,
1254 name,
1255 timeout=None,
1256 sleep=0.1,
1257 blocking=True,
1258 blocking_timeout=None,
1259 lock_class=None,
1260 thread_local=True,
1261 raise_on_release_error: bool = True,
1262 ):
1263 """
1264 Return a new Lock object using key ``name`` that mimics
1265 the behavior of threading.Lock.
1267 If specified, ``timeout`` indicates a maximum life for the lock.
1268 By default, it will remain locked until release() is called.
1270 ``sleep`` indicates the amount of time to sleep per loop iteration
1271 when the lock is in blocking mode and another client is currently
1272 holding the lock.
1274 ``blocking`` indicates whether calling ``acquire`` should block until
1275 the lock has been acquired or to fail immediately, causing ``acquire``
1276 to return False and the lock not being acquired. Defaults to True.
1277 Note this value can be overridden by passing a ``blocking``
1278 argument to ``acquire``.
1280 ``blocking_timeout`` indicates the maximum amount of time in seconds to
1281 spend trying to acquire the lock. A value of ``None`` indicates
1282 continue trying forever. ``blocking_timeout`` can be specified as a
1283 float or integer, both representing the number of seconds to wait.
1285 ``lock_class`` forces the specified lock implementation. Note that as
1286 of redis-py 3.0, the only lock class we implement is ``Lock`` (which is
1287 a Lua-based lock). So, it's unlikely you'll need this parameter, unless
1288 you have created your own custom lock class.
1290 ``thread_local`` indicates whether the lock token is placed in
1291 thread-local storage. By default, the token is placed in thread local
1292 storage so that a thread only sees its token, not a token set by
1293 another thread. Consider the following timeline:
1295 time: 0, thread-1 acquires `my-lock`, with a timeout of 5 seconds.
1296 thread-1 sets the token to "abc"
1297 time: 1, thread-2 blocks trying to acquire `my-lock` using the
1298 Lock instance.
1299 time: 5, thread-1 has not yet completed. redis expires the lock
1300 key.
1301 time: 5, thread-2 acquired `my-lock` now that it's available.
1302 thread-2 sets the token to "xyz"
1303 time: 6, thread-1 finishes its work and calls release(). if the
1304 token is *not* stored in thread local storage, then
1305 thread-1 would see the token value as "xyz" and would be
1306 able to successfully release the thread-2's lock.
1308 ``raise_on_release_error`` indicates whether to raise an exception when
1309 the lock is no longer owned when exiting the context manager. By default,
1310 this is True, meaning an exception will be raised. If False, the warning
1311 will be logged and the exception will be suppressed.
1313 In some use cases it's necessary to disable thread local storage. For
1314 example, if you have code where one thread acquires a lock and passes
1315 that lock instance to a worker thread to release later. If thread
1316 local storage isn't disabled in this case, the worker thread won't see
1317 the token set by the thread that acquired the lock. Our assumption
1318 is that these cases aren't common and as such default to using
1319 thread local storage."""
1320 if lock_class is None:
1321 lock_class = Lock
1322 return lock_class(
1323 self,
1324 name,
1325 timeout=timeout,
1326 sleep=sleep,
1327 blocking=blocking,
1328 blocking_timeout=blocking_timeout,
1329 thread_local=thread_local,
1330 raise_on_release_error=raise_on_release_error,
1331 )
1333 def set_response_callback(self, command, callback):
1334 """Set a custom Response Callback"""
1335 self.cluster_response_callbacks[command] = callback
1337 def _determine_nodes(
1338 self, *args, request_policy: RequestPolicy, **kwargs
1339 ) -> List["ClusterNode"]:
1340 """
1341 Determines a nodes the command should be executed on.
1342 """
1343 command = args[0].upper()
1344 if len(args) >= 2 and f"{args[0]} {args[1]}".upper() in self.command_flags:
1345 command = f"{args[0]} {args[1]}".upper()
1347 nodes_flag = kwargs.pop("nodes_flag", None)
1348 if nodes_flag is not None:
1349 # nodes flag passed by the user
1350 command_flag = nodes_flag
1351 else:
1352 # get the nodes group for this command if it was predefined
1353 command_flag = self.command_flags.get(command)
1355 if command_flag in self._command_flags_mapping:
1356 request_policy = self._command_flags_mapping[command_flag]
1358 policy_callback = self._policies_callback_mapping[request_policy]
1360 if request_policy == RequestPolicy.DEFAULT_KEYED:
1361 nodes = policy_callback(command, *args)
1362 elif request_policy == RequestPolicy.MULTI_SHARD:
1363 nodes = policy_callback(*args, **kwargs)
1364 elif request_policy == RequestPolicy.DEFAULT_KEYLESS:
1365 nodes = policy_callback(args[0])
1366 else:
1367 nodes = policy_callback()
1369 if args[0].lower() == "ft.aggregate":
1370 self._aggregate_nodes = nodes
1372 return nodes
1374 def _should_reinitialized(self):
1375 # To reinitialize the cluster on every MOVED error,
1376 # set reinitialize_steps to 1.
1377 # To avoid reinitializing the cluster on moved errors, set
1378 # reinitialize_steps to 0.
1379 if self.reinitialize_steps == 0:
1380 return False
1381 else:
1382 return self.reinitialize_counter % self.reinitialize_steps == 0
1384 def keyslot(self, key):
1385 """
1386 Calculate keyslot for a given key.
1387 See Keys distribution model in https://redis.io/topics/cluster-spec
1388 """
1389 k = self.encoder.encode(key)
1390 return key_slot(k)
1392 # HIMPORT orchestration. PREPARE/DISCARD/DISCARDALL mutate the one shared
1393 # HImportRegistry exactly once (every node pool references the same object, so the
1394 # change is visible cluster-wide and applied lazily per node). SET routes by key
1395 # slot to the owning primary and reuses that node's standalone himport_set (lazy
1396 # PREPARE bundled with SET). See ``.agents/himport_client_support_spec.md``.
1398 @property
1399 def himport_registry(self) -> HImportRegistry:
1400 """The cluster-wide HIMPORT fieldset registry (empty if none was declared).
1402 Read-only: the registry is mutated only through the HIMPORT command methods.
1403 """
1404 return self._himport_registry
1406 @experimental_method()
1407 def himport_prepare(self, fieldset_name: str, fields: Iterable[FieldT]) -> bool:
1408 """Declare an HIMPORT fieldset cluster-wide (shared registry, applied lazily)."""
1409 self._himport_registry.prepare(fieldset_name, fields)
1410 return True
1412 @experimental_method()
1413 def himport_discard(self, fieldset_name: str) -> int:
1414 """Remove an HIMPORT fieldset cluster-wide (shared registry, applied lazily)."""
1415 return 1 if self._himport_registry.discard(fieldset_name) else 0
1417 @experimental_method()
1418 def himport_discard_all(self) -> int:
1419 """Remove all HIMPORT fieldsets cluster-wide (shared registry, applied lazily)."""
1420 return self._himport_registry.discard_all()
1422 def _get_command_keys(self, *args):
1423 """
1424 Get the keys in the command. If the command has no keys in in, None is
1425 returned.
1427 NOTE: Due to a bug in redis<7.0, this function does not work properly
1428 for EVAL or EVALSHA when the `numkeys` arg is 0.
1429 - issue: https://github.com/redis/redis/issues/9493
1430 - fix: https://github.com/redis/redis/pull/9733
1432 So, don't use this function with EVAL or EVALSHA.
1433 """
1434 redis_conn = self.get_default_node().redis_connection
1435 return self.commands_parser.get_keys(redis_conn, *args)
1437 def determine_slot(self, *args) -> Optional[int]:
1438 """
1439 Figure out what slot to use based on args.
1441 Raises a RedisClusterException if there's a missing key and we can't
1442 determine what slots to map the command to; or, if the keys don't
1443 all map to the same key slot.
1444 """
1445 command = args[0]
1446 if self.command_flags.get(command) == SLOT_ID:
1447 # The command contains the slot ID
1448 return args[1]
1450 # Get the keys in the command
1452 # CLIENT TRACKING is a special case.
1453 # It doesn't have any keys, it needs to be sent to the provided nodes
1454 # By default it will be sent to all nodes.
1455 if command.upper() == "CLIENT TRACKING":
1456 return None
1458 # EVAL and EVALSHA are common enough that it's wasteful to go to the
1459 # redis server to parse the keys. Besides, there is a bug in redis<7.0
1460 # where `self._get_command_keys()` fails anyway. So, we special case
1461 # EVAL/EVALSHA.
1462 if command.upper() in ("EVAL", "EVALSHA"):
1463 # command syntax: EVAL "script body" num_keys ...
1464 if len(args) <= 2:
1465 raise RedisClusterException(f"Invalid args in command: {args}")
1466 num_actual_keys = int(args[2])
1467 eval_keys = args[3 : 3 + num_actual_keys]
1468 # if there are 0 keys, that means the script can be run on any node
1469 # so we can just return a random slot
1470 if len(eval_keys) == 0:
1471 return random.randrange(0, REDIS_CLUSTER_HASH_SLOTS)
1472 keys = eval_keys
1473 else:
1474 keys = self._get_command_keys(*args)
1475 if keys is None or len(keys) == 0:
1476 # FCALL can call a function with 0 keys, that means the function
1477 # can be run on any node so we can just return a random slot
1478 if command.upper() in ("FCALL", "FCALL_RO"):
1479 return random.randrange(0, REDIS_CLUSTER_HASH_SLOTS)
1480 raise RedisClusterException(
1481 "No way to dispatch this command to Redis Cluster. "
1482 "Missing key.\nYou can execute the command by specifying "
1483 f"target nodes.\nCommand: {args}"
1484 )
1486 # single key command
1487 if len(keys) == 1:
1488 return self.keyslot(keys[0])
1490 # multi-key command; we need to make sure all keys are mapped to
1491 # the same slot
1492 slots = {self.keyslot(key) for key in keys}
1493 if len(slots) != 1:
1494 raise RedisClusterException(
1495 f"{command} - all keys must map to the same key slot"
1496 )
1498 return slots.pop()
1500 def get_encoder(self):
1501 """
1502 Get the connections' encoder
1503 """
1504 return self.encoder
1506 def get_connection_kwargs(self):
1507 """
1508 Get the connections' key-word arguments
1509 """
1510 return self.nodes_manager.connection_kwargs
1512 def _is_nodes_flag(self, target_nodes):
1513 return isinstance(target_nodes, str) and target_nodes in self.node_flags
1515 def _parse_target_nodes(self, target_nodes):
1516 if isinstance(target_nodes, list):
1517 nodes = target_nodes
1518 elif isinstance(target_nodes, ClusterNode):
1519 # Supports passing a single ClusterNode as a variable
1520 nodes = [target_nodes]
1521 elif isinstance(target_nodes, dict):
1522 # Supports dictionaries of the format {node_name: node}.
1523 # It enables to execute commands with multi nodes as follows:
1524 # rc.cluster_save_config(rc.get_primaries())
1525 nodes = target_nodes.values()
1526 else:
1527 raise TypeError(
1528 "target_nodes type can be one of the following: "
1529 "node_flag (PRIMARIES, REPLICAS, RANDOM, ALL_NODES),"
1530 "ClusterNode, list<ClusterNode>, or dict<any, ClusterNode>. "
1531 f"The passed type is {type(target_nodes)}"
1532 )
1533 return nodes
1535 def execute_command(self, *args, **kwargs):
1536 return self._internal_execute_command(*args, **kwargs)
1538 def _internal_execute_command(self, *args, **kwargs):
1539 """
1540 Wrapper for ERRORS_ALLOW_RETRY error handling.
1542 It will try the number of times specified by the retries property from
1543 config option "self.retry" which defaults to 10 unless manually
1544 configured.
1546 If it reaches the number of times, the command will raise the exception
1548 Key argument :target_nodes: can be passed with the following types:
1549 nodes_flag: PRIMARIES, REPLICAS, ALL_NODES, RANDOM
1550 ClusterNode
1551 list<ClusterNode>
1552 dict<Any, ClusterNode>
1553 """
1554 target_nodes_specified = False
1555 is_default_node = False
1556 target_nodes = None
1557 passed_targets = kwargs.pop("target_nodes", None)
1558 command_policies = self._policy_resolver.resolve(args[0].lower())
1560 if passed_targets is not None and not self._is_nodes_flag(passed_targets):
1561 target_nodes = self._parse_target_nodes(passed_targets)
1562 target_nodes_specified = True
1564 if not command_policies and not target_nodes_specified:
1565 command = args[0].upper()
1566 if len(args) >= 2 and f"{args[0]} {args[1]}".upper() in self.command_flags:
1567 command = f"{args[0]} {args[1]}".upper()
1569 # We only could resolve key properties if command is not
1570 # in a list of pre-defined request policies
1571 command_flag = self.command_flags.get(command)
1572 if not command_flag:
1573 # Fallback to default policy
1574 if not self.get_default_node():
1575 slot = None
1576 else:
1577 slot = self.determine_slot(*args)
1578 if slot is None:
1579 command_policies = CommandPolicies()
1580 else:
1581 command_policies = CommandPolicies(
1582 request_policy=RequestPolicy.DEFAULT_KEYED,
1583 response_policy=ResponsePolicy.DEFAULT_KEYED,
1584 )
1585 else:
1586 if command_flag in self._command_flags_mapping:
1587 command_policies = CommandPolicies(
1588 request_policy=self._command_flags_mapping[command_flag]
1589 )
1590 else:
1591 command_policies = CommandPolicies()
1592 elif not command_policies and target_nodes_specified:
1593 command_policies = CommandPolicies()
1595 # If an error that allows retrying was thrown, the nodes and slots
1596 # cache were reinitialized. We will retry executing the command with
1597 # the updated cluster setup only when the target nodes can be
1598 # determined again with the new cache tables. Therefore, when target
1599 # nodes were passed to this function, we cannot retry the command
1600 # execution since the nodes may not be valid anymore after the tables
1601 # were reinitialized. So in case of passed target nodes,
1602 # retry_attempts will be set to 0.
1603 retry_attempts = 0 if target_nodes_specified else self.retry.get_retries()
1604 # Add one for the first execution
1605 execute_attempts = 1 + retry_attempts
1606 failure_count = 0
1608 # Start timing for observability
1609 start_time = time.monotonic()
1611 for _ in range(execute_attempts):
1612 try:
1613 res = {}
1614 if not target_nodes_specified:
1615 # Determine the nodes to execute the command on
1616 target_nodes = self._determine_nodes(
1617 *args,
1618 request_policy=command_policies.request_policy,
1619 nodes_flag=passed_targets,
1620 )
1622 if not target_nodes:
1623 raise RedisClusterException(
1624 f"No targets were found to execute {args} command on"
1625 )
1626 if (
1627 len(target_nodes) == 1
1628 and target_nodes[0] == self.get_default_node()
1629 ):
1630 is_default_node = True
1631 for node in target_nodes:
1632 res[node.name] = self._execute_command(node, *args, **kwargs)
1634 if command_policies.response_policy == ResponsePolicy.ONE_SUCCEEDED:
1635 break
1637 # Return the processed result
1638 return self._process_result(
1639 args[0],
1640 res,
1641 response_policy=command_policies.response_policy,
1642 **kwargs,
1643 )
1644 except Exception as e:
1645 if retry_attempts > 0 and type(e) in self.__class__.ERRORS_ALLOW_RETRY:
1646 if is_default_node:
1647 # Replace the default cluster node
1648 self.replace_default_node()
1649 # The nodes and slots cache were reinitialized.
1650 # Try again with the new cluster setup.
1651 retry_attempts -= 1
1652 failure_count += 1
1654 if hasattr(e, "connection"):
1655 self._record_command_metric(
1656 command_name=args[0],
1657 duration_seconds=time.monotonic() - start_time,
1658 connection=e.connection,
1659 error=e,
1660 )
1662 self._record_error_metric(
1663 error=e,
1664 connection=e.connection,
1665 retry_attempts=failure_count,
1666 )
1667 continue
1668 else:
1669 # raise the exception
1670 if hasattr(e, "connection"):
1671 self._record_error_metric(
1672 error=e,
1673 connection=e.connection,
1674 retry_attempts=failure_count,
1675 is_internal=False,
1676 )
1677 raise e
1679 def _himport_reconcile_discards(self, redis_node, connection):
1680 """Delegate to the shared sync HIMPORT executor."""
1681 return _himport_exec.reconcile_discards(redis_node, connection)
1683 def _himport_prepare_and_set(
1684 self,
1685 redis_node,
1686 connection,
1687 key,
1688 fieldset_name,
1689 values,
1690 fieldset,
1691 asking: bool = False,
1692 ):
1693 """Delegate to the shared sync HIMPORT executor."""
1694 return _himport_exec.prepare_and_set(
1695 redis_node,
1696 connection,
1697 key,
1698 fieldset_name,
1699 values,
1700 fieldset,
1701 asking=asking,
1702 )
1704 def _himport_execute_set(
1705 self,
1706 redis_node,
1707 connection,
1708 key,
1709 fieldset_name,
1710 values,
1711 asking: bool = False,
1712 ):
1713 """Delegate to the shared sync HIMPORT executor."""
1714 return _himport_exec.execute_set(
1715 redis_node, connection, key, fieldset_name, values, asking=asking
1716 )
1718 def _execute_command(self, target_node, *args, **kwargs):
1719 """
1720 Send a command to a node in the cluster
1721 """
1722 command = args[0]
1723 redis_node = None
1724 connection = None
1725 redirect_addr = None
1726 asking = False
1727 moved = False
1728 ttl = int(self.RedisClusterRequestTTL)
1730 # Start timing for observability
1731 start_time = time.monotonic()
1733 while ttl > 0:
1734 ttl -= 1
1735 try:
1736 if asking:
1737 target_node = self.get_node(node_name=redirect_addr)
1738 elif moved:
1739 # MOVED occurred and the slots cache was updated,
1740 # refresh the target node
1741 slot = self.determine_slot(*args)
1742 target_node = self.nodes_manager.get_node_from_slot(
1743 slot,
1744 self.read_from_replicas and command in READ_COMMANDS,
1745 self.load_balancing_strategy
1746 if command in READ_COMMANDS
1747 else None,
1748 )
1749 moved = False
1751 redis_node = self.get_redis_connection(target_node)
1752 connection = get_connection(redis_node)
1753 himport_set = parse_himport_set_args(args)
1754 if asking and himport_set is None:
1755 connection.send_command("ASKING")
1756 redis_node.parse_response(connection, "ASKING", **kwargs)
1757 asking = False
1758 if himport_set is not None:
1759 # args == (HIMPORT_SET, key, fieldset_name, *values). A raw
1760 # ``execute_command`` with too few args falls through to the
1761 # normal send path below so the server returns its arity error
1762 # instead of a client-side IndexError.
1763 # The cluster
1764 # executor lazily PREPAREs the fieldset on this connection and
1765 # reconciles deferred DISCARDs, then SETs; it already applies the
1766 # HIMPORT SET response callback, so it bypasses the cluster callback
1767 # block below.
1768 # This per-command branch in the hot dispatch path is deliberate
1769 # and has no cleaner alternative: this is the only seam where the
1770 # concrete routed connection is known, and connection-scoped
1771 # session setup can only happen once that connection is chosen.
1772 # On an ASK redirect ``asking`` is folded into the SET's own packed
1773 # write (see the guard above that suppresses the standalone ASKING
1774 # for HIMPORT SET) so the allowance sits immediately before the SET.
1775 # Clear ``asking`` first and carry the allowance in a dedicated
1776 # local: ``_himport_execute_set`` can raise a retriable MOVED/TRYAGAIN
1777 # mid-exchange, and a stale ``asking`` would shadow the moved-retry
1778 # branch on the next loop iteration (mirrors the async client).
1779 key, fieldset_name, values = himport_set
1780 ask_himport = asking
1781 asking = False
1782 response = self._himport_execute_set(
1783 redis_node,
1784 connection,
1785 key,
1786 fieldset_name,
1787 values,
1788 asking=ask_himport,
1789 )
1790 kwargs.pop("keys", None)
1791 else:
1792 connection.send_command(*args, **kwargs)
1793 response = redis_node.parse_response(connection, command, **kwargs)
1795 # Remove keys entry, it needs only for cache.
1796 kwargs.pop("keys", None)
1798 if command in self.cluster_response_callbacks:
1799 response = self.cluster_response_callbacks[command](
1800 response, **kwargs
1801 )
1803 self._record_command_metric(
1804 command_name=command,
1805 duration_seconds=time.monotonic() - start_time,
1806 connection=connection,
1807 )
1808 return response
1809 except AuthenticationError as e:
1810 e.connection = connection if connection is not None else target_node
1811 self._record_command_metric(
1812 command_name=command,
1813 duration_seconds=time.monotonic() - start_time,
1814 connection=e.connection,
1815 error=e,
1816 )
1817 raise
1818 except MaxConnectionsError as e:
1819 # MaxConnectionsError indicates client-side resource exhaustion
1820 # (too many connections in the pool), not a node failure.
1821 # Don't treat this as a node failure - just re-raise the error
1822 # without reinitializing the cluster.
1823 # The connection in the error is used to report the metrics based on host and port info
1824 # so we use the target node object which contains the host and port info
1825 # because we did not get the connection yet
1826 e.connection = target_node
1827 self._record_command_metric(
1828 command_name=command,
1829 duration_seconds=time.monotonic() - start_time,
1830 connection=e.connection,
1831 error=e,
1832 )
1833 raise
1834 except (ConnectionError, TimeoutError) as e:
1835 if is_debug_log_enabled():
1836 socket_address = self._extracts_socket_address(connection)
1837 args_log_str = truncate_text(" ".join(map(safe_str, args)))
1838 logger.debug(
1839 f"{type(e).__name__} received for command {args_log_str}, on node {target_node.name}, "
1840 f"and connection: {connection} using local socket address: {socket_address}, error: {e}"
1841 )
1842 # this is used to report the metrics based on host and port info
1843 e.connection = connection if connection else target_node
1845 # ConnectionError can also be raised if we couldn't get a
1846 # connection from the pool before timing out, so check that
1847 # this is an actual connection before attempting to disconnect.
1848 if connection is not None:
1849 connection.disconnect()
1851 # Instead of setting to None, properly handle the pool
1852 # Get the pool safely - redis_connection could be set to None
1853 # by another thread between the check and access
1854 redis_conn = target_node.redis_connection
1855 if redis_conn is not None:
1856 pool = redis_conn.connection_pool
1857 if pool is not None:
1858 with pool._lock:
1859 # take care for the active connections in the pool
1860 pool.update_active_connections_for_reconnect()
1861 # disconnect all free connections
1862 pool.disconnect_free_connections()
1864 # Move the failed node to the end of the cached nodes list
1865 self.nodes_manager.move_node_to_end_of_cached_nodes(target_node.name)
1867 # DON'T set redis_connection = None - keep the pool for reuse
1868 # provide the name of the failed node so we can try it last
1869 self.nodes_manager.initialize(last_failed_node_name=target_node.name)
1870 self._record_command_metric(
1871 command_name=command,
1872 duration_seconds=time.monotonic() - start_time,
1873 connection=e.connection,
1874 error=e,
1875 )
1876 raise e
1877 except MovedError as e:
1878 if is_debug_log_enabled():
1879 socket_address = self._extracts_socket_address(connection)
1880 args_log_str = truncate_text(" ".join(map(safe_str, args)))
1881 logger.debug(
1882 f"MOVED error received for command {args_log_str}, on node {target_node.name}, "
1883 f"and connection: {connection} using local socket address: {socket_address}, error: {e}"
1884 )
1885 # First, we will try to patch the slots/nodes cache with the
1886 # redirected node output and try again. If MovedError exceeds
1887 # 'reinitialize_steps' number of times, we will force
1888 # reinitializing the tables, and then try again.
1889 # 'reinitialize_steps' counter will increase faster when
1890 # the same client object is shared between multiple threads. To
1891 # reduce the frequency you can set this variable in the
1892 # RedisCluster constructor.
1893 self.reinitialize_counter += 1
1894 if self._should_reinitialized():
1895 # during this call all connections are closed or marked for disconnect,
1896 # so we don't need to disconnect the changed node's connections
1897 self.nodes_manager.initialize(
1898 additional_startup_nodes_info=[(e.host, e.port)]
1899 )
1900 # Reset the counter
1901 self.reinitialize_counter = 0
1902 else:
1903 self.nodes_manager.move_slot(e)
1904 moved = True
1905 self._record_command_metric(
1906 command_name=command,
1907 duration_seconds=time.monotonic() - start_time,
1908 connection=connection,
1909 error=e,
1910 )
1911 self._record_error_metric(
1912 error=e,
1913 connection=connection,
1914 )
1915 except TryAgainError as e:
1916 if is_debug_log_enabled():
1917 socket_address = self._extracts_socket_address(connection)
1918 args_log_str = truncate_text(" ".join(map(safe_str, args)))
1919 logger.debug(
1920 f"TRYAGAIN error received for command {args_log_str}, on node {target_node.name}, "
1921 f"and connection: {connection} using local socket address: {socket_address}"
1922 )
1923 if ttl < self.RedisClusterRequestTTL / 2:
1924 time.sleep(0.05)
1926 self._record_command_metric(
1927 command_name=command,
1928 duration_seconds=time.monotonic() - start_time,
1929 connection=connection,
1930 error=e,
1931 )
1932 self._record_error_metric(
1933 error=e,
1934 connection=connection,
1935 )
1936 except AskError as e:
1937 if is_debug_log_enabled():
1938 socket_address = self._extracts_socket_address(connection)
1939 args_log_str = truncate_text(" ".join(map(safe_str, args)))
1940 logger.debug(
1941 f"ASK error received for command {args_log_str}, on node {target_node.name}, "
1942 f"and connection: {connection} using local socket address: {socket_address}, error: {e}"
1943 )
1944 redirect_addr = get_node_name(host=e.host, port=e.port)
1945 asking = True
1947 self._record_command_metric(
1948 command_name=command,
1949 duration_seconds=time.monotonic() - start_time,
1950 connection=connection,
1951 error=e,
1952 )
1953 self._record_error_metric(
1954 error=e,
1955 connection=connection,
1956 )
1957 except (ClusterDownError, SlotNotCoveredError) as e:
1958 # ClusterDownError can occur during a failover and to get
1959 # self-healed, we will try to reinitialize the cluster layout
1960 # and retry executing the command
1962 # SlotNotCoveredError can occur when the cluster is not fully
1963 # initialized or can be temporary issue.
1964 # We will try to reinitialize the cluster topology
1965 # and retry executing the command
1967 time.sleep(0.25)
1968 self.nodes_manager.initialize()
1970 # if we have a connection, use it, otherwise use the target node
1971 # object which contains the host and port info
1972 # this is used to report the metrics based on host and port info
1973 e.connection = connection if connection else target_node
1974 self._record_command_metric(
1975 command_name=command,
1976 duration_seconds=time.monotonic() - start_time,
1977 connection=e.connection,
1978 error=e,
1979 )
1980 raise
1981 except ResponseError as e:
1982 # this is used to report the metrics based on host and port info
1983 # ResponseError typically happens after get_connection() succeeds,
1984 # so connection should be available
1985 e.connection = connection if connection else target_node
1986 self._record_command_metric(
1987 command_name=command,
1988 duration_seconds=time.monotonic() - start_time,
1989 connection=e.connection,
1990 error=e,
1991 )
1992 raise
1993 except Exception as e:
1994 if connection:
1995 connection.disconnect()
1997 # if we have a connection, use it, otherwise use the target node
1998 # object which contains the host and port info
1999 # this is used to report the metrics based on host and port info
2000 e.connection = connection if connection else target_node
2001 self._record_command_metric(
2002 command_name=command,
2003 duration_seconds=time.monotonic() - start_time,
2004 connection=e.connection,
2005 error=e,
2006 )
2007 raise e
2008 finally:
2009 if connection is not None:
2010 redis_node.connection_pool.release(connection)
2012 e = ClusterError("TTL exhausted.")
2013 # In this case we should have an active connection.
2014 # If we are here, we have received many MOVED or ASK errors and finally exhausted the TTL.
2015 # This means that we used an active connection to read from the socket.
2016 # This is used to report metrics based on the host and port information.
2017 e.connection = connection
2018 self._record_command_metric(
2019 command_name=command,
2020 duration_seconds=time.monotonic() - start_time,
2021 connection=connection,
2022 error=e,
2023 )
2024 raise e
2026 def _record_command_metric(
2027 self,
2028 command_name: str,
2029 duration_seconds: float,
2030 connection: Connection,
2031 error=None,
2032 ):
2033 """
2034 Records operation duration metric directly.
2035 """
2036 host = connection.host if connection else "unknown"
2037 port = connection.port if connection else 0
2038 db = str(connection.db) if connection and hasattr(connection, "db") else "0"
2040 record_operation_duration(
2041 command_name=command_name,
2042 duration_seconds=duration_seconds,
2043 server_address=host,
2044 server_port=port,
2045 db_namespace=db,
2046 error=error,
2047 )
2049 def _record_error_metric(
2050 self,
2051 error: Exception,
2052 connection: Connection,
2053 is_internal: bool = True,
2054 retry_attempts: Optional[int] = None,
2055 ):
2056 """
2057 Records error count metric directly.
2058 """
2059 record_error_count(
2060 server_address=connection.host,
2061 server_port=connection.port,
2062 network_peer_address=connection.host,
2063 network_peer_port=connection.port,
2064 error_type=error,
2065 retry_attempts=retry_attempts if retry_attempts is not None else 0,
2066 is_internal=is_internal,
2067 )
2069 def _extracts_socket_address(
2070 self, connection: Optional[Connection]
2071 ) -> Optional[int]:
2072 if connection is None:
2073 return None
2074 try:
2075 socket_address = (
2076 connection._sock.getsockname() if connection._sock else None
2077 )
2078 socket_address = socket_address[1] if socket_address else None
2079 except (AttributeError, OSError):
2080 pass
2081 return socket_address
2083 def close(self) -> None:
2084 try:
2085 with self._lock:
2086 if self.nodes_manager:
2087 self.nodes_manager.close()
2088 except AttributeError:
2089 # RedisCluster's __init__ can fail before nodes_manager is set
2090 pass
2092 def _process_result(self, command, res, response_policy: ResponsePolicy, **kwargs):
2093 """
2094 Process the result of the executed command.
2095 The function would return a dict or a single value.
2097 :type command: str
2098 :type res: dict
2100 `res` should be in the following format:
2101 Dict<node_name, command_result>
2102 """
2103 if command in self.result_callbacks:
2104 res = self.result_callbacks[command](command, res, **kwargs)
2105 elif len(res) == 1:
2106 # When we execute the command on a single node, we can
2107 # remove the dictionary and return a single response
2108 res = list(res.values())[0]
2110 return self._policies_callback_mapping[response_policy](res)
2112 def load_external_module(self, funcname, func):
2113 """
2114 This function can be used to add externally defined redis modules,
2115 and their namespaces to the redis client.
2117 ``funcname`` - A string containing the name of the function to create
2118 ``func`` - The function, being added to this class.
2119 """
2120 setattr(self, funcname, func)
2122 def transaction(self, func, *watches, **kwargs):
2123 """
2124 Convenience method for executing the callable `func` as a transaction
2125 while watching all keys specified in `watches`. The 'func' callable
2126 should expect a single argument which is a Pipeline object.
2127 """
2128 shard_hint = kwargs.pop("shard_hint", None)
2129 value_from_callable = kwargs.pop("value_from_callable", False)
2130 watch_delay = kwargs.pop("watch_delay", None)
2131 with self.pipeline(True, shard_hint) as pipe:
2132 while True:
2133 try:
2134 if watches:
2135 pipe.watch(*watches)
2136 func_value = func(pipe)
2137 exec_value = pipe.execute()
2138 return func_value if value_from_callable else exec_value
2139 except WatchError:
2140 if watch_delay is not None and watch_delay > 0:
2141 time.sleep(watch_delay)
2142 continue
2145class ClusterNode:
2146 def __init__(self, host, port, server_type=None, redis_connection=None):
2147 if host == "localhost":
2148 host = socket.gethostbyname(host)
2150 self.host = host
2151 self.port = port
2152 self.name = get_node_name(host, port)
2153 self.server_type = server_type
2154 self.redis_connection = redis_connection
2156 def __repr__(self):
2157 return (
2158 f"[host={self.host},"
2159 f"port={self.port},"
2160 f"name={self.name},"
2161 f"server_type={self.server_type},"
2162 f"redis_connection={self.redis_connection}]"
2163 )
2165 def __eq__(self, obj):
2166 return isinstance(obj, ClusterNode) and obj.name == self.name
2168 def __hash__(self):
2169 return hash(self.name)
2172class LoadBalancingStrategy(Enum):
2173 ROUND_ROBIN = "round_robin"
2174 ROUND_ROBIN_REPLICAS = "round_robin_replicas"
2175 RANDOM = "random"
2176 RANDOM_REPLICA = "random_replica"
2179class LoadBalancer:
2180 """
2181 Round-Robin Load Balancing
2182 """
2184 def __init__(self, start_index: int = 0) -> None:
2185 self.primary_to_idx: dict[str, int] = {}
2186 self.start_index: int = start_index
2187 self._lock: threading.Lock = threading.Lock()
2189 def get_server_index(
2190 self,
2191 primary: str,
2192 list_size: int,
2193 load_balancing_strategy: LoadBalancingStrategy = LoadBalancingStrategy.ROUND_ROBIN,
2194 ) -> int:
2195 if load_balancing_strategy == LoadBalancingStrategy.RANDOM_REPLICA:
2196 return self._get_random_server_index(
2197 list_size,
2198 replicas_only=True,
2199 )
2200 elif load_balancing_strategy == LoadBalancingStrategy.RANDOM:
2201 return self._get_random_server_index(
2202 list_size,
2203 replicas_only=False,
2204 )
2205 else:
2206 return self._get_round_robin_index(
2207 primary,
2208 list_size,
2209 load_balancing_strategy == LoadBalancingStrategy.ROUND_ROBIN_REPLICAS,
2210 )
2212 def reset(self) -> None:
2213 with self._lock:
2214 self.primary_to_idx.clear()
2216 def _get_random_server_index(self, list_size: int, replicas_only: bool) -> int:
2217 return random.randint(1 if replicas_only else 0, list_size - 1)
2219 def _get_round_robin_index(
2220 self, primary: str, list_size: int, replicas_only: bool
2221 ) -> int:
2222 with self._lock:
2223 server_index = self.primary_to_idx.setdefault(primary, self.start_index)
2224 if replicas_only and server_index == 0:
2225 # skip the primary node index
2226 server_index = 1
2227 # Update the index for the next round
2228 self.primary_to_idx[primary] = (server_index + 1) % list_size
2229 return server_index
2232class NodesManager:
2233 def __init__(
2234 self,
2235 startup_nodes: list[ClusterNode],
2236 from_url=False,
2237 require_full_coverage=False,
2238 lock: Optional[threading.RLock] = None,
2239 dynamic_startup_nodes=True,
2240 connection_pool_class=ConnectionPool,
2241 address_remap: Optional[Callable[[Tuple[str, int]], Tuple[str, int]]] = None,
2242 cache: Optional[CacheInterface] = None,
2243 cache_config: Optional[CacheConfig] = None,
2244 cache_factory: Optional[CacheFactoryInterface] = None,
2245 event_dispatcher: Optional[EventDispatcher] = None,
2246 maint_notifications_config: Optional[MaintNotificationsConfig] = None,
2247 himport_registry: HImportRegistry | None = None,
2248 **kwargs,
2249 ):
2250 # Shared, cluster-wide HIMPORT registry object, injected onto every node's pool
2251 # in create_redis_node (not forwarded through connection_kwargs, so all nodes
2252 # reuse the one object rather than rebuilding it per node).
2253 self.himport_registry = himport_registry
2254 self.nodes_cache: dict[str, ClusterNode] = {}
2255 self.slots_cache: dict[int, list[ClusterNode]] = {}
2256 self.startup_nodes: dict[str, ClusterNode] = {n.name: n for n in startup_nodes}
2257 self.default_node: Optional[ClusterNode] = None
2258 self._epoch: int = 0
2259 self.from_url = from_url
2260 self._require_full_coverage = require_full_coverage
2261 self._dynamic_startup_nodes = dynamic_startup_nodes
2262 self.connection_pool_class = connection_pool_class
2263 self.address_remap = address_remap
2264 self._cache: Optional[CacheInterface] = None
2265 if cache:
2266 self._cache = cache
2267 elif cache_factory is not None:
2268 self._cache = cache_factory.get_cache()
2269 elif cache_config is not None:
2270 self._cache = CacheFactory(cache_config).get_cache()
2271 self.connection_kwargs = kwargs
2272 self.read_load_balancer = LoadBalancer()
2274 # nodes_cache / slots_cache / startup_nodes / default_node are protected by _lock
2275 if lock is None:
2276 self._lock = threading.RLock()
2277 else:
2278 self._lock = lock
2280 # initialize holds _initialization_lock to dedup multiple calls to reinitialize;
2281 # note that if we hold both _lock and _initialization_lock, we _must_ acquire
2282 # _initialization_lock first (ie: to have a consistent order) to avoid deadlock.
2283 self._initialization_lock: threading.RLock = threading.RLock()
2285 if event_dispatcher is None:
2286 self._event_dispatcher = EventDispatcher()
2287 else:
2288 self._event_dispatcher = event_dispatcher
2289 self._credential_provider = self.connection_kwargs.get(
2290 "credential_provider", None
2291 )
2292 self.maint_notifications_config = maint_notifications_config
2294 self.initialize()
2296 def get_node(
2297 self,
2298 host: Optional[str] = None,
2299 port: Optional[int] = None,
2300 node_name: Optional[str] = None,
2301 ) -> Optional[ClusterNode]:
2302 """
2303 Get the requested node from the cluster's nodes.
2304 nodes.
2305 :return: ClusterNode if the node exists, else None
2306 """
2307 if host and port:
2308 # the user passed host and port
2309 if host == "localhost":
2310 host = socket.gethostbyname(host)
2311 with self._lock:
2312 return self.nodes_cache.get(get_node_name(host=host, port=port))
2313 elif node_name:
2314 with self._lock:
2315 return self.nodes_cache.get(node_name)
2316 else:
2317 return None
2319 def move_slot(self, e: Union[AskError, MovedError]):
2320 """
2321 Update the slot's node with the redirected one
2322 """
2323 node_changed = False
2324 with self._lock:
2325 redirected_node = self.get_node(host=e.host, port=e.port)
2326 if redirected_node is not None:
2327 # The node already exists
2328 if redirected_node.server_type is not PRIMARY:
2329 # Update the node's server type
2330 redirected_node.server_type = PRIMARY
2331 else:
2332 # This is a new node, we will add it to the nodes cache
2333 redirected_node = ClusterNode(e.host, e.port, PRIMARY)
2334 self.nodes_cache[redirected_node.name] = redirected_node
2336 slot_nodes = self.slots_cache[e.slot_id]
2337 if redirected_node not in slot_nodes:
2338 # The new slot owner is a new server, or a server from a different
2339 # shard. We need to remove all current nodes from the slot's list
2340 # (including replications) and add just the new node.
2341 self.slots_cache[e.slot_id] = [redirected_node]
2342 node_changed = True
2343 elif redirected_node is not slot_nodes[0]:
2344 # The MOVED error resulted from a failover, and the new slot owner
2345 # had previously been a replica.
2346 old_primary = slot_nodes[0]
2347 # Update the old primary to be a replica and add it to the end of
2348 # the slot's node list
2349 old_primary.server_type = REPLICA
2350 slot_nodes.append(old_primary)
2351 # Remove the old replica, which is now a primary, from the slot's
2352 # node list
2353 slot_nodes.remove(redirected_node)
2354 # Override the old primary with the new one
2355 slot_nodes[0] = redirected_node
2356 if self.default_node == old_primary:
2357 # Update the default node with the new primary
2358 self.default_node = redirected_node
2359 node_changed = True
2360 # else: circular MOVED to current primary -> no-op
2361 # Dispatch outside the lock so listeners can acquire their own locks
2362 # without risk of deadlock. Skipped on the no-op branch to avoid
2363 # needless reconciliation walks under MOVED storms. A listener must
2364 # not break slots-cache refresh; log and continue so a single buggy
2365 # listener cannot starve the rest.
2366 if node_changed:
2367 try:
2368 self._event_dispatcher.dispatch(AfterSlotsCacheRefreshEvent())
2369 except Exception as exc:
2370 # Don't shadow the method parameter ``e``: ``except as`` binds
2371 # the listener exception in the function scope and ``del``s
2372 # the name on block exit (PEP 3134), which would also wipe
2373 # out the original AskError/MovedError parameter.
2374 logger.exception(
2375 "listener raised during slots-cache refresh: %s: %s",
2376 type(exc).__name__,
2377 exc,
2378 )
2380 @deprecated_args(
2381 args_to_warn=["server_type"],
2382 reason=(
2383 "In case you need select some load balancing strategy "
2384 "that will use replicas, please set it through 'load_balancing_strategy'"
2385 ),
2386 version="5.3.0",
2387 )
2388 def get_node_from_slot(
2389 self,
2390 slot: int,
2391 read_from_replicas: bool = False,
2392 load_balancing_strategy: Optional[LoadBalancingStrategy] = None,
2393 server_type: Optional[Literal["primary", "replica"]] = None,
2394 ) -> ClusterNode:
2395 """
2396 Gets a node that servers this hash slot
2397 """
2399 if read_from_replicas is True and load_balancing_strategy is None:
2400 load_balancing_strategy = LoadBalancingStrategy.ROUND_ROBIN
2402 with self._lock:
2403 if self.slots_cache.get(slot) is None or len(self.slots_cache[slot]) == 0:
2404 raise SlotNotCoveredError(
2405 f'Slot "{slot}" not covered by the cluster. '
2406 + f'"require_full_coverage={self._require_full_coverage}"'
2407 )
2409 if len(self.slots_cache[slot]) > 1 and load_balancing_strategy:
2410 # get the server index using the strategy defined in load_balancing_strategy
2411 primary_name = self.slots_cache[slot][0].name
2412 node_idx = self.read_load_balancer.get_server_index(
2413 primary_name, len(self.slots_cache[slot]), load_balancing_strategy
2414 )
2415 elif (
2416 server_type is None
2417 or server_type == PRIMARY
2418 or len(self.slots_cache[slot]) == 1
2419 ):
2420 # return a primary
2421 node_idx = 0
2422 else:
2423 # return a replica
2424 # randomly choose one of the replicas
2425 node_idx = random.randint(1, len(self.slots_cache[slot]) - 1)
2427 return self.slots_cache[slot][node_idx]
2429 def get_nodes_by_server_type(self, server_type: Literal["primary", "replica"]):
2430 """
2431 Get all nodes with the specified server type
2432 :param server_type: 'primary' or 'replica'
2433 :return: list of ClusterNode
2434 """
2435 with self._lock:
2436 return [
2437 node
2438 for node in self.nodes_cache.values()
2439 if node.server_type == server_type
2440 ]
2442 @deprecated_function(
2443 reason="This method is not used anymore internally. The startup nodes are populated automatically.",
2444 version="7.0.2",
2445 )
2446 def populate_startup_nodes(self, nodes):
2447 """
2448 Populate all startup nodes and filters out any duplicates
2449 """
2450 with self._lock:
2451 for n in nodes:
2452 self.startup_nodes[n.name] = n
2454 def move_node_to_end_of_cached_nodes(self, node_name: str) -> None:
2455 """
2456 Move a failing node to the end of startup_nodes and nodes_cache so it's
2457 tried last during reinitialization and when selecting the default node.
2458 If the node is not in the respective list, nothing is done.
2459 """
2460 # Move in startup_nodes
2461 if node_name in self.startup_nodes and len(self.startup_nodes) > 1:
2462 node = self.startup_nodes.pop(node_name)
2463 self.startup_nodes[node_name] = node # Re-insert at end
2465 # Move in nodes_cache - this affects get_nodes_by_server_type ordering
2466 # which is used to select the default_node during initialize()
2467 if node_name in self.nodes_cache and len(self.nodes_cache) > 1:
2468 node = self.nodes_cache.pop(node_name)
2469 self.nodes_cache[node_name] = node # Re-insert at end
2471 def check_slots_coverage(self, slots_cache):
2472 # Validate if all slots are covered or if we should try next
2473 # startup node
2474 for i in range(0, REDIS_CLUSTER_HASH_SLOTS):
2475 if i not in slots_cache:
2476 return False
2477 return True
2479 def create_redis_connections(self, nodes):
2480 """
2481 This function will create a redis connection to all nodes in :nodes:
2482 """
2483 connection_pools = []
2484 for node in nodes:
2485 if node.redis_connection is None:
2486 node.redis_connection = self.create_redis_node(
2487 host=node.host,
2488 port=node.port,
2489 maint_notifications_config=self.maint_notifications_config,
2490 **self.connection_kwargs,
2491 )
2492 connection_pools.append(node.redis_connection.connection_pool)
2494 self._event_dispatcher.dispatch(
2495 AfterPooledConnectionsInstantiationEvent(
2496 connection_pools, ClientType.SYNC, self._credential_provider
2497 )
2498 )
2500 def create_redis_node(
2501 self,
2502 host,
2503 port,
2504 **kwargs,
2505 ):
2506 # We are configuring the connection pool not to retry
2507 # connections on lower level clients to avoid retrying
2508 # connections to nodes that are not reachable
2509 # and to avoid blocking the connection pool.
2510 # The only error that will have some handling in the lower
2511 # level clients is ConnectionError which will trigger disconnection
2512 # of the socket.
2513 # The retries will be handled on cluster client level
2514 # where we will have proper handling of the cluster topology
2515 node_retry_config = Retry(
2516 backoff=NoBackoff(), retries=0, supported_errors=(ConnectionError,)
2517 )
2519 if self.from_url:
2520 # Create a redis node with a custom connection pool
2521 kwargs.update({"host": host})
2522 kwargs.update({"port": port})
2523 kwargs.update({"cache": self._cache})
2524 kwargs.update({"retry": node_retry_config})
2525 r = Redis(connection_pool=self.connection_pool_class(**kwargs))
2526 else:
2527 r = Redis(
2528 host=host,
2529 port=port,
2530 cache=self._cache,
2531 retry=node_retry_config,
2532 **kwargs,
2533 )
2534 # Share the one cluster-wide HIMPORT registry with this node's pool. Injected
2535 # here (rather than forwarded via connection_kwargs) so every node reuses the
2536 # same object; the node has no connections yet, so this is safe.
2537 if self.himport_registry is not None:
2538 r.connection_pool.himport_registry = self.himport_registry
2539 r.connection_pool.connection_kwargs["himport_registry"] = (
2540 self.himport_registry
2541 )
2542 return r
2544 def _get_or_create_cluster_node(self, host, port, role, tmp_nodes_cache):
2545 node_name = get_node_name(host, port)
2546 # check if we already have this node in the tmp_nodes_cache
2547 target_node = tmp_nodes_cache.get(node_name)
2548 if target_node is None:
2549 # before creating a new cluster node, check if the cluster node already
2550 # exists in the current nodes cache and has a valid connection so we can
2551 # reuse it
2552 redis_connection: Optional[Redis] = None
2553 with self._lock:
2554 previous_node = self.nodes_cache.get(node_name)
2555 if previous_node:
2556 redis_connection = previous_node.redis_connection
2557 # don't update the old ClusterNode, so we don't update its role
2558 # outside of the lock
2559 target_node = ClusterNode(host, port, role, redis_connection)
2560 # add this node to the nodes cache
2561 tmp_nodes_cache[target_node.name] = target_node
2563 return target_node
2565 def _get_epoch(self) -> int:
2566 """
2567 Get the current epoch value. This method exists primarily to allow
2568 tests to mock the epoch fetch and control race condition timing.
2569 """
2570 with self._lock:
2571 return self._epoch
2573 def initialize(
2574 self,
2575 additional_startup_nodes_info: Optional[List[Tuple[str, int]]] = None,
2576 disconnect_startup_nodes_pools: bool = True,
2577 last_failed_node_name: Optional[str] = None,
2578 ):
2579 """
2580 Initializes the nodes cache, slots cache and redis connections.
2581 :startup_nodes:
2582 Responsible for discovering other nodes in the cluster
2583 :disconnect_startup_nodes_pools:
2584 Whether to disconnect the connection pool of the startup nodes
2585 after the initialization is complete. This is useful when the
2586 startup nodes are not part of the cluster and we want to avoid
2587 keeping the connection open.
2588 :additional_startup_nodes_info:
2589 Additional nodes to add temporarily to the startup nodes.
2590 The additional nodes will be used just in the process of extraction of the slots
2591 and nodes information from the cluster.
2592 This is useful when we want to add new nodes to the cluster
2593 and initialize the client
2594 with them.
2595 The format of the list is a list of tuples, where each tuple contains
2596 the host and port of the node.
2597 :last_failed_node_name:
2598 Name of the node that just failed and should be tried only after
2599 other startup and additional startup nodes during this refresh.
2600 """
2601 self.reset()
2602 tmp_nodes_cache = {}
2603 tmp_slots = {}
2604 disagreements = []
2605 startup_nodes_reachable = False
2606 fully_covered = False
2607 kwargs = self.connection_kwargs
2608 exception = None
2609 epoch = self._get_epoch()
2610 if additional_startup_nodes_info is None:
2611 additional_startup_nodes_info = []
2613 with self._initialization_lock:
2614 with self._lock:
2615 if epoch != self._epoch:
2616 # another thread has already re-initialized the nodes; don't
2617 # bother running again
2618 return
2620 with self._lock:
2621 startup_nodes = list(self.startup_nodes.values())
2622 deferred_failed_nodes = []
2623 if last_failed_node_name is not None:
2624 for index, node in enumerate(startup_nodes):
2625 if node.name == last_failed_node_name:
2626 deferred_failed_nodes.append(startup_nodes.pop(index))
2627 break
2628 if len(startup_nodes) > 1:
2629 # Vary which startup node is queried first so clients do not
2630 # all reinitialize through the same node.
2631 random.shuffle(startup_nodes)
2633 additional_startup_nodes = [
2634 ClusterNode(host, port) for host, port in additional_startup_nodes_info
2635 ]
2636 if last_failed_node_name is not None:
2637 for index, node in enumerate(additional_startup_nodes):
2638 if node.name == last_failed_node_name:
2639 if not deferred_failed_nodes:
2640 deferred_failed_nodes.append(node)
2641 additional_startup_nodes.pop(index)
2642 break
2643 if is_debug_log_enabled():
2644 logger.debug(
2645 f"Topology refresh: using additional nodes: {[node.name for node in additional_startup_nodes]}; "
2646 f"and startup nodes: {[node.name for node in startup_nodes]}"
2647 )
2649 for startup_node in chain(
2650 startup_nodes,
2651 additional_startup_nodes,
2652 deferred_failed_nodes,
2653 ):
2654 try:
2655 if startup_node.redis_connection:
2656 r = startup_node.redis_connection
2658 else:
2659 # Create a new Redis connection
2660 if is_debug_log_enabled():
2661 socket_timeout = kwargs.get("socket_timeout", "not set")
2662 socket_connect_timeout = kwargs.get(
2663 "socket_connect_timeout", "not set"
2664 )
2665 maint_enabled = (
2666 self.maint_notifications_config.enabled
2667 if self.maint_notifications_config
2668 else False
2669 )
2670 logger.debug(
2671 "Topology refresh: Creating new Redis connection to "
2672 f"{startup_node.host}:{startup_node.port}; "
2673 f"with socket_timeout: {socket_timeout}, and "
2674 f"socket_connect_timeout: {socket_connect_timeout}, "
2675 "and maint_notifications enabled: "
2676 f"{maint_enabled}"
2677 )
2678 r = self.create_redis_node(
2679 startup_node.host,
2680 startup_node.port,
2681 maint_notifications_config=self.maint_notifications_config,
2682 **kwargs,
2683 )
2684 if startup_node in self.startup_nodes.values():
2685 self.startup_nodes[startup_node.name].redis_connection = r
2686 else:
2687 startup_node.redis_connection = r
2688 try:
2689 # Make sure cluster mode is enabled on this node
2690 cluster_slots = str_if_bytes(r.execute_command("CLUSTER SLOTS"))
2691 if disconnect_startup_nodes_pools:
2692 with r.connection_pool._lock:
2693 # take care to clear connections before we move on
2694 # mark all active connections for reconnect - they will be
2695 # reconnected on next use, but will allow current in flight commands to complete first
2696 r.connection_pool.update_active_connections_for_reconnect()
2697 # Needed to clear READONLY state when it is no longer applicable
2698 r.connection_pool.disconnect_free_connections()
2699 except ResponseError:
2700 raise RedisClusterException(
2701 "Cluster mode is not enabled on this node"
2702 )
2703 startup_nodes_reachable = True
2704 except Exception as e:
2705 # Try the next startup node.
2706 # The exception is saved and raised only if we have no more nodes.
2707 exception = e
2708 continue
2710 # CLUSTER SLOTS command results in the following output:
2711 # [[slot_section[from_slot,to_slot,master,replica1,...,replicaN]]]
2712 # where each node contains the following list: [IP, port, node_id]
2713 # Therefore, cluster_slots[0][2][0] will be the IP address of the
2714 # primary node of the first slot section.
2715 # If there's only one server in the cluster, its ``host`` is ''
2716 # Fix it to the host in startup_nodes
2717 if (
2718 len(cluster_slots) == 1
2719 and len(cluster_slots[0][2][0]) == 0
2720 and len(self.startup_nodes) == 1
2721 ):
2722 cluster_slots[0][2][0] = startup_node.host
2724 for slot in cluster_slots:
2725 primary_node = slot[2]
2726 host = str_if_bytes(primary_node[0])
2727 if host == "":
2728 host = startup_node.host
2729 port = int(primary_node[1])
2730 host, port = self.remap_host_port(host, port)
2732 nodes_for_slot = []
2734 target_node = self._get_or_create_cluster_node(
2735 host, port, PRIMARY, tmp_nodes_cache
2736 )
2737 nodes_for_slot.append(target_node)
2739 replica_nodes = slot[3:]
2740 for replica_node in replica_nodes:
2741 host = str_if_bytes(replica_node[0])
2742 port = int(replica_node[1])
2743 host, port = self.remap_host_port(host, port)
2744 target_replica_node = self._get_or_create_cluster_node(
2745 host, port, REPLICA, tmp_nodes_cache
2746 )
2747 nodes_for_slot.append(target_replica_node)
2749 for i in range(int(slot[0]), int(slot[1]) + 1):
2750 if i not in tmp_slots:
2751 tmp_slots[i] = nodes_for_slot
2752 else:
2753 # Validate that 2 nodes want to use the same slot cache
2754 # setup
2755 tmp_slot = tmp_slots[i][0]
2756 if tmp_slot.name != target_node.name:
2757 disagreements.append(
2758 f"{tmp_slot.name} vs {target_node.name} on slot: {i}"
2759 )
2761 if len(disagreements) > 5:
2762 raise RedisClusterException(
2763 f"startup_nodes could not agree on a valid "
2764 f"slots cache: {', '.join(disagreements)}"
2765 )
2767 fully_covered = self.check_slots_coverage(tmp_slots)
2768 if fully_covered:
2769 # Don't need to continue to the next startup node if all
2770 # slots are covered
2771 break
2773 if not startup_nodes_reachable:
2774 raise RedisClusterException(
2775 f"Redis Cluster cannot be connected. Please provide at least "
2776 f"one reachable node: {str(exception)}"
2777 ) from exception
2779 # Create Redis connections to all nodes
2780 self.create_redis_connections(list(tmp_nodes_cache.values()))
2782 # Check if the slots are not fully covered
2783 if not fully_covered and self._require_full_coverage:
2784 # Despite the requirement that the slots be covered, there
2785 # isn't a full coverage
2786 raise RedisClusterException(
2787 f"All slots are not covered after query all startup_nodes. "
2788 f"{len(tmp_slots)} of {REDIS_CLUSTER_HASH_SLOTS} "
2789 f"covered..."
2790 )
2792 # Set the tmp variables to the real variables
2793 with self._lock:
2794 self.nodes_cache = tmp_nodes_cache
2795 self.slots_cache = tmp_slots
2796 # Set the default node
2797 self.default_node = self.get_nodes_by_server_type(PRIMARY)[0]
2798 if self._dynamic_startup_nodes:
2799 # Populate the startup nodes with all discovered nodes
2800 self.startup_nodes = tmp_nodes_cache
2801 # Increment the epoch to signal that initialization has completed
2802 self._epoch += 1
2803 # Dispatch so listeners (e.g. ClusterPubSub) can reconcile per-node
2804 # state after slot ownership may have changed. A listener must not
2805 # break slots-cache refresh; log and continue so a single buggy
2806 # listener cannot starve the rest.
2807 try:
2808 self._event_dispatcher.dispatch(AfterSlotsCacheRefreshEvent())
2809 except Exception as e:
2810 logger.exception(
2811 "listener raised during slots-cache refresh: %s: %s",
2812 type(e).__name__,
2813 e,
2814 )
2816 def close(self) -> None:
2817 with self._lock:
2818 self.default_node = None
2819 nodes = tuple(self.nodes_cache.values())
2820 for node in nodes:
2821 if node.redis_connection:
2822 node.redis_connection.close()
2824 def reset(self):
2825 try:
2826 self.read_load_balancer.reset()
2827 except TypeError:
2828 # The read_load_balancer is None, do nothing
2829 pass
2831 def remap_host_port(self, host: str, port: int) -> Tuple[str, int]:
2832 """
2833 Remap the host and port returned from the cluster to a different
2834 internal value. Useful if the client is not connecting directly
2835 to the cluster.
2836 """
2837 if self.address_remap:
2838 return self.address_remap((host, port))
2839 return host, port
2841 def find_connection_owner(self, connection: Connection) -> Optional[ClusterNode]:
2842 node_name = get_node_name(connection.host, connection.port)
2843 with self._lock:
2844 for node in tuple(self.nodes_cache.values()):
2845 if node.redis_connection:
2846 conn_args = node.redis_connection.connection_pool.connection_kwargs
2847 if node_name == get_node_name(
2848 conn_args.get("host"), conn_args.get("port")
2849 ):
2850 return node
2851 return None
2854def _unregister_slots_cache_listener(
2855 dispatcher_ref: "weakref.ref[EventDispatcher]",
2856 listener: EventListenerInterface,
2857 event_type: Type[object],
2858) -> None:
2859 # Module-level finalizer callback. Kept free of strong references to the
2860 # owning ClusterPubSub so attaching it via weakref.finalize does not
2861 # extend the pubsub's lifetime.
2862 dispatcher = dispatcher_ref()
2863 if dispatcher is not None:
2864 dispatcher.unregister_listeners({event_type: [listener]})
2867class ClusterPubSubSlotsCacheListener(EventListenerInterface):
2868 """
2869 Listener that forwards AfterSlotsCacheRefreshEvent to a ClusterPubSub.
2871 Holds a weak reference to the pubsub so it does not keep the instance
2872 alive. Deterministic cleanup of the dispatcher's strong reference to this
2873 listener is performed by a ``weakref.finalize`` attached to the owning
2874 ClusterPubSub in ``ClusterPubSub.__init__``.
2875 """
2877 def __init__(self, pubsub: "ClusterPubSub") -> None:
2878 self._pubsub_ref: "weakref.ref[ClusterPubSub]" = weakref.ref(pubsub)
2880 def listen(self, event: object) -> None:
2881 pubsub = self._pubsub_ref()
2882 if pubsub is None:
2883 # Race window between pubsub GC and the finalizer running; safe
2884 # no-op, finalizer will remove this listener shortly.
2885 return
2886 try:
2887 pubsub.on_slots_changed()
2888 except Exception as e:
2889 # Listeners must not break slots-cache refresh; log and continue so
2890 # a single buggy pubsub cannot starve the rest.
2891 logger.exception(
2892 "pubsub %r raised during slots-cache change: %s: %s",
2893 pubsub,
2894 type(e).__name__,
2895 e,
2896 )
2899class ClusterPubSub(PubSub):
2900 """
2901 Wrapper for PubSub class.
2903 IMPORTANT: before using ClusterPubSub, read about the known limitations
2904 with pubsub in Cluster mode and learn how to workaround them:
2905 https://redis.readthedocs.io/en/stable/clustering.html#known-pubsub-limitations
2906 """
2908 def __init__(
2909 self,
2910 redis_cluster,
2911 node=None,
2912 host=None,
2913 port=None,
2914 push_handler_func=None,
2915 event_dispatcher: Optional["EventDispatcher"] = None,
2916 **kwargs,
2917 ):
2918 """
2919 When a pubsub instance is created without specifying a node, a single
2920 node will be transparently chosen for the pubsub connection on the
2921 first command execution. The node will be determined by:
2922 1. Hashing the channel name in the request to find its keyslot
2923 2. Selecting a node that handles the keyslot: If read_from_replicas is
2924 set to true or load_balancing_strategy is set, a replica can be selected.
2926 :type redis_cluster: RedisCluster
2927 :type node: ClusterNode
2928 :type host: str
2929 :type port: int
2930 """
2931 self.node = None
2932 self.set_pubsub_node(redis_cluster, node, host, port)
2933 connection_pool = (
2934 None
2935 if self.node is None
2936 else redis_cluster.get_redis_connection(self.node).connection_pool
2937 )
2938 self.cluster = redis_cluster
2939 self.node_pubsub_mapping = {}
2940 # Reverse index: shard channel (normalized) -> owning node.name. Used to
2941 # route sunsubscribe calls and reconcile subscriptions after slot
2942 # migration / failover.
2943 self._shard_channel_to_node: dict = {}
2944 # Dedicated lock for shard-subscription bookkeeping. Distinct from
2945 # PubSub.self._lock (which serializes wire I/O on the cluster-level
2946 # connection used by aclose / send_command / regular subscribe) so
2947 # that reconciliation cannot starve those unrelated paths during
2948 # long per-channel migrations.
2949 self._shard_state_lock: threading.RLock = threading.RLock()
2950 # Worker executor for off-loading slot-migration reconciliation from
2951 # the dispatch call site (mirrors async's asyncio.create_task model so
2952 # the thread that triggered MovedError / topology refresh is not
2953 # blocked on per-channel sunsubscribe / ssubscribe network I/O).
2954 # Lazy-created on first on_slots_changed() to avoid a persistent
2955 # worker thread for pubsubs that never see a slot migration.
2956 # Initialized before super().__init__() because PubSub.__init__ calls
2957 # self.reset(), which resolves to ClusterPubSub.reset() and reads
2958 # these attributes.
2959 self._reconcile_executor: Optional[ThreadPoolExecutor] = None
2960 # In-flight reconciliation futures; tracked so reset() can cancel
2961 # pending work and so exceptions surface via a done-callback.
2962 self._reconcile_futures: Set[Future] = set()
2963 self._pubsubs_generator = self._pubsubs_generator()
2964 if event_dispatcher is None:
2965 self._event_dispatcher = EventDispatcher()
2966 else:
2967 self._event_dispatcher = event_dispatcher
2968 super().__init__(
2969 connection_pool=connection_pool,
2970 encoder=redis_cluster.encoder,
2971 push_handler_func=push_handler_func,
2972 event_dispatcher=self._event_dispatcher,
2973 **kwargs,
2974 )
2975 # Subscribe to slots-cache change notifications so shard subscriptions
2976 # can be reconciled automatically after topology refreshes.
2977 nm_dispatcher = redis_cluster.nodes_manager._event_dispatcher
2978 self._slots_cache_listener = ClusterPubSubSlotsCacheListener(self)
2979 nm_dispatcher.register_listeners(
2980 {AfterSlotsCacheRefreshEvent: [self._slots_cache_listener]}
2981 )
2982 # Deterministic GC-time cleanup so short-lived pubsubs do not leak
2983 # listeners in the dispatcher when no slots-refresh event ever fires.
2984 weakref.finalize(
2985 self,
2986 _unregister_slots_cache_listener,
2987 weakref.ref(nm_dispatcher),
2988 self._slots_cache_listener,
2989 AfterSlotsCacheRefreshEvent,
2990 )
2992 def set_pubsub_node(self, cluster, node=None, host=None, port=None):
2993 """
2994 The pubsub node will be set according to the passed node, host and port
2995 When none of the node, host, or port are specified - the node is set
2996 to None and will be determined by the keyslot of the channel in the
2997 first command to be executed.
2998 RedisClusterException will be thrown if the passed node does not exist
2999 in the cluster.
3000 If host is passed without port, or vice versa, a DataError will be
3001 thrown.
3002 :type cluster: RedisCluster
3003 :type node: ClusterNode
3004 :type host: str
3005 :type port: int
3006 """
3007 if node is not None:
3008 # node is passed by the user
3009 self._raise_on_invalid_node(cluster, node, node.host, node.port)
3010 pubsub_node = node
3011 elif host is not None and port is not None:
3012 # host and port passed by the user
3013 node = cluster.get_node(host=host, port=port)
3014 self._raise_on_invalid_node(cluster, node, host, port)
3015 pubsub_node = node
3016 elif any([host, port]) is True:
3017 # only 'host' or 'port' passed
3018 raise DataError("Passing a host requires passing a port, and vice versa")
3019 else:
3020 # nothing passed by the user. set node to None
3021 pubsub_node = None
3023 self.node = pubsub_node
3025 def get_pubsub_node(self):
3026 """
3027 Get the node that is being used as the pubsub connection
3028 """
3029 return self.node
3031 def _raise_on_invalid_node(self, redis_cluster, node, host, port):
3032 """
3033 Raise a RedisClusterException if the node is None or doesn't exist in
3034 the cluster.
3035 """
3036 if node is None or redis_cluster.get_node(node_name=node.name) is None:
3037 raise RedisClusterException(
3038 f"Node {host}:{port} doesn't exist in the cluster"
3039 )
3041 def execute_command(self, *args):
3042 """
3043 Execute a subscribe/unsubscribe command.
3045 Taken code from redis-py and tweak to make it work within a cluster.
3046 """
3047 # NOTE: don't parse the response in this function -- it could pull a
3048 # legitimate message off the stack if the connection is already
3049 # subscribed to one or more channels
3051 if self.connection is None:
3052 if self.connection_pool is None:
3053 if len(args) > 1:
3054 # Hash the first channel and get one of the nodes holding
3055 # this slot
3056 channel = args[1]
3057 slot = self.cluster.keyslot(channel)
3058 node = self.cluster.nodes_manager.get_node_from_slot(
3059 slot,
3060 self.cluster.read_from_replicas,
3061 self.cluster.load_balancing_strategy,
3062 )
3063 else:
3064 # Get a random node
3065 node = self.cluster.get_random_node()
3066 self.node = node
3067 redis_connection = self.cluster.get_redis_connection(node)
3068 self.connection_pool = redis_connection.connection_pool
3069 self.connection = self.connection_pool.get_connection()
3070 # register a callback that re-subscribes to any channels we
3071 # were listening to when we were disconnected
3072 self.connection.register_connect_callback(self.on_connect)
3073 if self.push_handler_func is not None:
3074 self.connection._parser.set_pubsub_push_handler(self.push_handler_func)
3075 self._event_dispatcher.dispatch(
3076 AfterPubSubConnectionInstantiationEvent(
3077 self.connection, self.connection_pool, ClientType.SYNC, self._lock
3078 )
3079 )
3080 connection = self.connection
3081 self._execute(connection, connection.send_command, *args)
3083 def _resubscribe_shard_channels(self) -> None:
3084 # A single node can own multiple slot ranges, so a batched
3085 # ``SSUBSCRIBE`` covering every tracked channel would be rejected by
3086 # Redis with a ``CROSSSLOT`` error. Group by hash slot and emit one
3087 # ``SSUBSCRIBE`` per slot.
3088 by_slot: defaultdict[int, dict] = defaultdict(dict)
3089 for k, v in self.shard_channels.items():
3090 by_slot[key_slot(self.encoder.encode(k))][k] = v
3091 for subscriptions in by_slot.values():
3092 self._resubscribe(subscriptions, self.ssubscribe)
3094 def _get_node_pubsub(self, node):
3095 try:
3096 return self.node_pubsub_mapping[node.name]
3097 except KeyError:
3098 redis_connection = self.cluster.get_redis_connection(node)
3099 pubsub = redis_connection.pubsub(
3100 push_handler_func=self.push_handler_func,
3101 )
3102 # Replay shard subscriptions on reconnect with slot-aware grouping
3103 # so that channels spanning multiple slots owned by this node do
3104 # not trigger a CROSSSLOT error.
3105 pubsub._resubscribe_shard_channels = MethodType(
3106 ClusterPubSub._resubscribe_shard_channels, pubsub
3107 )
3108 self.node_pubsub_mapping[node.name] = pubsub
3109 return pubsub
3111 def _find_node_name_for_pubsub(self, pubsub):
3112 for node_name, node_pubsub in self.node_pubsub_mapping.items():
3113 if node_pubsub is pubsub:
3114 return node_name
3115 return None
3117 def _sharded_message_generator(self, timeout=0.0):
3118 for _ in range(len(self.node_pubsub_mapping)):
3119 pubsub = next(self._pubsubs_generator)
3120 # Don't pass ignore_subscribe_messages here - let get_sharded_message
3121 # handle the filtering after processing subscription state changes
3122 message = pubsub.get_message(
3123 ignore_subscribe_messages=False, timeout=timeout
3124 )
3125 if message is not None:
3126 return pubsub, message
3127 return None, None
3129 def _pubsubs_generator(self):
3130 while True:
3131 current_nodes = list(self.node_pubsub_mapping.values())
3132 if not current_nodes:
3133 return # Avoid infinite loop when no subscriptions exist
3134 yield from current_nodes
3136 def get_sharded_message(
3137 self, ignore_subscribe_messages=False, timeout=0.0, target_node=None
3138 ):
3139 if target_node:
3140 # Use .get(): migration-driven cleanup in the sunsubscribe branch
3141 # below and reset() both remove entries from node_pubsub_mapping,
3142 # so a caller polling with target_node may race the cleanup. Match
3143 # the async counterpart's None-handling rather than raising
3144 # KeyError. None pubsub falls through to "no message available".
3145 pubsub = self.node_pubsub_mapping.get(target_node.name)
3146 if pubsub is not None:
3147 # Don't pass ignore_subscribe_messages here - let get_sharded_message
3148 # handle the filtering after processing subscription state changes
3149 message = pubsub.get_message(
3150 ignore_subscribe_messages=False, timeout=timeout
3151 )
3152 else:
3153 message = None
3154 else:
3155 pubsub, message = self._sharded_message_generator(timeout=timeout)
3156 if message is None:
3157 return None
3158 # Only sunsubscribe mutates cluster-level shard state; bypassing the
3159 # lock on the data-message hot path keeps smessage delivery from
3160 # competing with the reconciliation worker for _shard_state_lock.
3161 if str_if_bytes(message["type"]) == "sunsubscribe":
3162 # Serialize state mutation against reinitialize_shard_subscriptions
3163 # (worker thread). The blocking get_message above intentionally
3164 # runs outside the lock so reconciliation is not stalled by long
3165 # polls.
3166 with self._shard_state_lock:
3167 if message["channel"] in self.pending_unsubscribe_shard_channels:
3168 # User-initiated sunsubscribe: drop from cluster-level tracking.
3169 self.pending_unsubscribe_shard_channels.remove(message["channel"])
3170 self.shard_channels.pop(message["channel"], None)
3171 self._shard_channel_to_node.pop(message["channel"], None)
3172 # Drop the per-node pubsub that delivered the confirmation once
3173 # it no longer holds any shard subscriptions, regardless of
3174 # whether the sunsubscribe was user-initiated or driven by
3175 # slot-migration reconciliation (_migrate_shard_channel, which
3176 # intentionally does not add the channel to
3177 # pending_unsubscribe_shard_channels). This releases the
3178 # dedicated connection that would otherwise linger.
3179 # Identifying the receiving pubsub directly (rather than via
3180 # the cluster's current slot map) is required after slot
3181 # migration, where the channel's owner is no longer the node
3182 # that received our original SSUBSCRIBE.
3183 if pubsub is not None and not pubsub.subscribed:
3184 name = self._find_node_name_for_pubsub(pubsub)
3185 if name is not None:
3186 try:
3187 pubsub.reset()
3188 except Exception:
3189 pass
3190 self.node_pubsub_mapping.pop(name, None)
3191 # Mirror PubSub.handle_message: the empty-check belongs in the
3192 # unsubscribe branch since that is the only path that can
3193 # reduce shard_channels here.
3194 if not self.channels and not self.patterns and not self.shard_channels:
3195 self.subscribed_event.clear()
3196 # Only suppress subscribe/unsubscribe messages, not data messages (smessage)
3197 if str_if_bytes(message["type"]) in ("ssubscribe", "sunsubscribe"):
3198 if self.ignore_subscribe_messages or ignore_subscribe_messages:
3199 return None
3200 return message
3202 def ssubscribe(
3203 self, *args: ChannelT | Subscription, **kwargs: PubSubHandler
3204 ) -> None:
3205 """
3206 Subscribe to shard channels.
3208 Channels supplied as keyword arguments expect a channel name as the key
3209 and a callable as the value. ``Subscription`` objects can also be
3210 supplied positionally with an optional handler.
3211 """
3212 s_channels = parse_pubsub_subscriptions(args, kwargs)
3213 # Serialize against reinitialize_shard_subscriptions (worker thread)
3214 # so the reverse index, shard_channels, and node_pubsub_mapping are
3215 # not mutated concurrently.
3216 with self._shard_state_lock:
3217 for s_channel, handler in s_channels.items():
3218 node = self.cluster.get_node_from_key(s_channel)
3219 if not node:
3220 continue
3221 # Lazy re-route: if this channel is already tracked against a
3222 # different node (e.g. after a slot migration), migrate it now
3223 # so the caller's intent is applied on the current owner.
3224 normalized_key = next(iter(self._normalize_keys({s_channel: None})))
3225 old_name = self._shard_channel_to_node.get(normalized_key)
3226 if old_name and old_name != node.name:
3227 # Match PubSub.ssubscribe() dict.update() semantics: the
3228 # caller's newly supplied handler (including None) always
3229 # overrides any previously registered handler.
3230 self._migrate_shard_channel(
3231 normalized_key,
3232 handler,
3233 old_name,
3234 node,
3235 )
3236 continue
3237 pubsub = self._get_node_pubsub(node)
3238 if handler:
3239 pubsub.ssubscribe(Subscription(s_channel, handler))
3240 else:
3241 pubsub.ssubscribe(s_channel)
3242 self.shard_channels.update(pubsub.shard_channels)
3243 self._shard_channel_to_node[normalized_key] = node.name
3244 self.pending_unsubscribe_shard_channels.difference_update(
3245 self._normalize_keys({s_channel: None})
3246 )
3247 if pubsub.subscribed and not self.subscribed:
3248 self.subscribed_event.set()
3249 self.health_check_response_counter = 0
3251 def sunsubscribe(self, *args):
3252 if args:
3253 args = list_or_args(args[0], args[1:])
3254 else:
3255 args = list(self.shard_channels)
3257 # Serialize against reinitialize_shard_subscriptions: the reverse
3258 # index and node_pubsub_mapping must not change between the lookup
3259 # and the per-node sunsubscribe call below.
3260 with self._shard_state_lock:
3261 for s_channel in args:
3262 normalized_key = next(iter(self._normalize_keys({s_channel: None})))
3263 # Route via the reverse index so we unsubscribe on the node
3264 # that actually holds the subscription. After a slot migration
3265 # the cluster's current owner may no longer be that node.
3266 name = self._shard_channel_to_node.get(normalized_key)
3267 if name and name in self.node_pubsub_mapping:
3268 p = self.node_pubsub_mapping[name]
3269 else:
3270 node = self.cluster.get_node_from_key(s_channel)
3271 if not node or node.name not in self.node_pubsub_mapping:
3272 continue
3273 p = self.node_pubsub_mapping[node.name]
3274 p.sunsubscribe(s_channel)
3275 self.pending_unsubscribe_shard_channels.update(
3276 p.pending_unsubscribe_shard_channels
3277 )
3279 def reinitialize_shard_subscriptions(self):
3280 """
3281 Reconcile per-node shard subscriptions against the cluster's current
3282 slot ownership map. For each tracked shard channel whose owning node
3283 has changed (e.g. after CLUSTER SETSLOT / failover), sunsubscribe on
3284 the old node's pubsub and ssubscribe on the new owner's pubsub,
3285 preserving any registered handler.
3286 """
3287 uncovered: list = []
3288 made_progress = False
3289 first_migrate_error: Optional[BaseException] = None
3290 with self._shard_state_lock:
3291 for channel, handler in list(self.shard_channels.items()):
3292 try:
3293 new_node = self.cluster.get_node_from_key(channel)
3294 except SlotNotCoveredError:
3295 # Slot is transiently uncovered (mid-migration / partial
3296 # topology refresh). Defer this channel so coverable
3297 # siblings still reconcile this pass; we surface the
3298 # error below so the caller (and logs) know not every
3299 # channel was reconciled. Retry happens on the next
3300 # slots-cache change notification.
3301 uncovered.append(channel)
3302 continue
3303 old_name = self._shard_channel_to_node.get(channel)
3304 if old_name == new_node.name:
3305 continue
3306 try:
3307 self._migrate_shard_channel(channel, handler, old_name, new_node)
3308 made_progress = True
3309 except (ConnectionError, TimeoutError, OSError) as e:
3310 # Transient connectivity error while subscribing on the
3311 # new owner (or unsubscribing on the old owner if its
3312 # handler chose to re-raise). Do not abort reconciliation
3313 # for sibling channels: _shard_channel_to_node was not
3314 # advanced for this channel, so the next slots-cache
3315 # change notification will retry it.
3316 logger.warning(
3317 "shard channel %r migration deferred: %s: %s",
3318 channel,
3319 type(e).__name__,
3320 e,
3321 )
3322 if first_migrate_error is None:
3323 first_migrate_error = e
3324 continue
3325 # Garbage-collect per-node pubsubs that no longer hold any
3326 # subscription so their connections are released.
3327 for name, pubsub in list(self.node_pubsub_mapping.items()):
3328 if not pubsub.subscribed:
3329 try:
3330 pubsub.reset()
3331 except Exception:
3332 pass
3333 self.node_pubsub_mapping.pop(name, None)
3334 if uncovered:
3335 # Surface the uncovered channels so the caller (and observer
3336 # notification path) knows reconciliation was incomplete. All
3337 # coverable siblings have already been migrated above.
3338 raise SlotNotCoveredError(
3339 f"{len(uncovered)} shard channel(s) left unreconciled; "
3340 f"slot(s) not covered by the cluster: {uncovered!r}"
3341 )
3342 if first_migrate_error is not None and not made_progress:
3343 # Every migration attempted in this pass failed transiently and
3344 # nothing else made progress. Re-raise the first caught error
3345 # (typically the root cause; later failures are often downstream
3346 # symptoms of the same unreachable node) so the worker's done-
3347 # callback surfaces a single representative failure through the
3348 # same logger channel used for SlotNotCoveredError. Per-channel
3349 # WARNINGs above preserve the full forensic detail.
3350 raise first_migrate_error
3352 def _migrate_shard_channel(self, channel, handler, old_name, new_node):
3353 # Detach from the old per-node pubsub, best-effort: the old node may
3354 # already be unreachable during migration / failover.
3355 if old_name and old_name in self.node_pubsub_mapping:
3356 old_pubsub = self.node_pubsub_mapping[old_name]
3357 try:
3358 old_pubsub.sunsubscribe(channel)
3359 except (ConnectionError, TimeoutError, OSError):
3360 # redis-py's Connection has already called ``disconnect()``
3361 # before raising (see Connection.read_response /
3362 # send_packed_command with ``disconnect_on_error=True``),
3363 # so ``old_pubsub``'s dedicated socket is gone. Two cases:
3364 #
3365 # 1. The old node is no longer in the cluster topology
3366 # (e.g. removed by failover / topology refresh): no
3367 # reconnect target exists, so ``old_pubsub.subscribed``
3368 # would stay True forever and the end-of-pass GC block
3369 # would skip it. Drop it eagerly so the round-robin
3370 # generator does not keep yielding a dead pubsub that
3371 # produces periodic errors from ``get_sharded_message``.
3372 # 2. The old node is still known (transiently slow /
3373 # unreachable): ``PubSub._execute`` auto-reconnects and
3374 # ``on_connect`` re-subscribes to remaining channels,
3375 # so other subscriptions on the same pubsub recover
3376 # naturally. Leave it alone.
3377 if self.cluster.get_node(node_name=old_name) is None:
3378 try:
3379 old_pubsub.reset()
3380 except Exception:
3381 pass
3382 self.node_pubsub_mapping.pop(old_name, None)
3383 # Attach to the new per-node pubsub, preserving the handler. Decode to
3384 # a text key only when we must pass it as a kwarg (handler present).
3385 new_pubsub = self._get_node_pubsub(new_node)
3386 if handler:
3387 new_pubsub.ssubscribe(Subscription(channel, handler))
3388 else:
3389 new_pubsub.ssubscribe(channel)
3390 self.shard_channels.update(new_pubsub.shard_channels)
3391 normalized_key = next(iter(self._normalize_keys({channel: None})))
3392 self._shard_channel_to_node[normalized_key] = new_node.name
3393 self.pending_unsubscribe_shard_channels.difference_update(
3394 self._normalize_keys({channel: None})
3395 )
3396 if new_pubsub.subscribed and not self.subscribed:
3397 self.subscribed_event.set()
3398 self.health_check_response_counter = 0
3400 def on_slots_changed(self):
3401 # Observer hook invoked by NodesManager after a slots-cache refresh.
3402 # Schedule reconciliation on a dedicated worker thread so the caller
3403 # (typically MovedError handling in _execute_command or the topology
3404 # refresh thread in initialize()) is not blocked on the network I/O
3405 # performed by reinitialize_shard_subscriptions. Mirrors the async
3406 # path's asyncio.create_task model. No-op when there are no shard
3407 # subscriptions to reconcile.
3408 if not self.shard_channels:
3409 return
3410 # Serialize lazy executor creation and submission against concurrent
3411 # on_slots_changed calls (EventDispatcher releases its lock before
3412 # invoking listeners, so two MovedError-handling threads can land
3413 # here at once) and against reset() which tears the executor down.
3414 # Without this, two threads could each create a ThreadPoolExecutor
3415 # and one would be orphaned (leaking its worker thread); a reset()
3416 # interleaved between create and submit() could also raise
3417 # RuntimeError("cannot schedule new futures after shutdown").
3418 with self._shard_state_lock:
3419 if self._reconcile_executor is None:
3420 self._reconcile_executor = ThreadPoolExecutor(
3421 max_workers=1,
3422 thread_name_prefix="redis-cluster-pubsub-reconcile",
3423 )
3424 future = self._reconcile_executor.submit(
3425 self.reinitialize_shard_subscriptions
3426 )
3427 self._reconcile_futures.add(future)
3428 future.add_done_callback(self._discard_reconcile_future)
3429 # Consume the future's exception (if any) so it is not silently lost.
3430 # reinitialize_shard_subscriptions surfaces SlotNotCoveredError when
3431 # a slot is still transiently uncovered; route it through the same
3432 # logger channel as the async path for consistent observability.
3433 future.add_done_callback(self._log_reconcile_future_exception)
3435 def _discard_reconcile_future(self, future: "Future") -> None:
3436 # Done-callback fires on the worker thread. Take _shard_state_lock so
3437 # the discard observes the same mutual-exclusion discipline as the
3438 # add() / clear() sites; without it the set mutation is correct only
3439 # because of CPython's GIL and would race under free-threaded builds.
3440 with self._shard_state_lock:
3441 self._reconcile_futures.discard(future)
3443 @staticmethod
3444 def _log_reconcile_future_exception(future: "Future") -> None:
3445 if future.cancelled():
3446 return
3447 exc = future.exception()
3448 if exc is not None:
3449 logger.error(
3450 "shard subscription reconciliation failed: %r", exc, exc_info=exc
3451 )
3453 def reset(self) -> None:
3454 # Hold _shard_state_lock across the entire teardown so it observes
3455 # the same mutual-exclusion discipline as ssubscribe / sunsubscribe /
3456 # get_sharded_message / reinitialize_shard_subscriptions, which all
3457 # mutate shard_channels, _shard_channel_to_node, and
3458 # node_pubsub_mapping under this lock. Without it, super().reset()
3459 # rebinds shard_channels and pending_unsubscribe_shard_channels in
3460 # parallel with a concurrent user-thread mutation, silently dropping
3461 # subscription intent. cancel_futures drops queued reconciliation
3462 # work; the currently-running task (if any) is already serialized
3463 # against us by this same lock - shutdown(wait=False) avoids waiting
3464 # on the worker thread's join, not on its critical section.
3465 with self._shard_state_lock:
3466 if self._reconcile_executor is not None:
3467 self._reconcile_executor.shutdown(wait=False, cancel_futures=True)
3468 self._reconcile_executor = None
3469 self._reconcile_futures.clear()
3470 # Tear down per-node pubsubs (parity with async aclose) so they
3471 # don't leak their dedicated connections and don't replay stale
3472 # shard_channels via PubSub.on_connect on a subsequent reconnect.
3473 # Errors are swallowed because reset() is also a fallback path
3474 # from __del__; we cannot let one buggy per-node pubsub mask the
3475 # rest of the teardown.
3476 for pubsub in self.node_pubsub_mapping.values():
3477 try:
3478 pubsub.reset()
3479 except Exception:
3480 pass
3481 # Drop the now-dead per-node pubsubs from the mapping so the
3482 # round-robin in _pubsubs_generator / _sharded_message_generator
3483 # cannot yield them between teardown and re-subscription.
3484 self.node_pubsub_mapping.clear()
3485 # _pubsubs_generator captures node_pubsub_mapping.values() into
3486 # a local list inside ``yield from``; clearing the mapping does
3487 # not reach references already held by that captured snapshot,
3488 # so a generator suspended mid-yield-from would still surface
3489 # the now-reset() per-node pubsubs after re-subscription.
3490 # Recreate it to drop the captured list. type(self) bypasses
3491 # the instance-level self-shadow established at __init__
3492 # (self._pubsubs_generator = self._pubsubs_generator()).
3493 self._pubsubs_generator = type(self)._pubsubs_generator(self)
3494 super().reset()
3495 self._shard_channel_to_node = {}
3497 def get_redis_connection(self):
3498 """
3499 Get the Redis connection of the pubsub connected node.
3500 """
3501 if self.node is not None:
3502 return self.node.redis_connection
3504 def disconnect(self):
3505 """
3506 Disconnect the pubsub connection.
3507 """
3508 if self.connection:
3509 self.connection.disconnect()
3510 for pubsub in self.node_pubsub_mapping.values():
3511 if pubsub.connection:
3512 pubsub.connection.disconnect()
3515class ClusterPipeline(RedisCluster):
3516 """
3517 Support for Redis pipeline
3518 in cluster mode
3519 """
3521 ERRORS_ALLOW_RETRY = (
3522 ConnectionError,
3523 TimeoutError,
3524 MovedError,
3525 AskError,
3526 TryAgainError,
3527 )
3529 NO_SLOTS_COMMANDS = {"UNWATCH"}
3530 IMMEDIATE_EXECUTE_COMMANDS = {"WATCH", "UNWATCH"}
3531 UNWATCH_COMMANDS = {"DISCARD", "EXEC", "UNWATCH"}
3533 @deprecated_args(
3534 args_to_warn=[
3535 "cluster_error_retry_attempts",
3536 ],
3537 reason="Please configure the 'retry' object instead",
3538 version="6.0.0",
3539 )
3540 def __init__(
3541 self,
3542 nodes_manager: "NodesManager",
3543 commands_parser: "CommandsParser",
3544 result_callbacks: Optional[Dict[str, Callable]] = None,
3545 cluster_response_callbacks: Optional[Dict[str, Callable]] = None,
3546 startup_nodes: Optional[List["ClusterNode"]] = None,
3547 read_from_replicas: bool = False,
3548 load_balancing_strategy: Optional[LoadBalancingStrategy] = None,
3549 cluster_error_retry_attempts: int = DEFAULT_RETRY_COUNT,
3550 reinitialize_steps: int = 5,
3551 retry: Optional[Retry] = None,
3552 lock=None,
3553 transaction=False,
3554 policy_resolver: PolicyResolver = StaticPolicyResolver(),
3555 event_dispatcher: Optional["EventDispatcher"] = None,
3556 **kwargs,
3557 ):
3558 """ """
3559 self.command_stack = []
3560 self.nodes_manager = nodes_manager
3561 # Share the parent cluster's HIMPORT registry (held on the NodesManager and
3562 # referenced by every node pool). The inherited himport_prepare/discard/
3563 # discard_all mutate this one object, so a fieldset declared on the pipeline is
3564 # visible to the batched himport_set pre-flight exactly as on the parent client.
3565 self._himport_registry = nodes_manager.himport_registry
3566 self.commands_parser = commands_parser
3567 self.refresh_table_asap = False
3568 self.result_callbacks = (
3569 result_callbacks or self.__class__.RESULT_CALLBACKS.copy()
3570 )
3571 self.startup_nodes = startup_nodes if startup_nodes else []
3572 self.read_from_replicas = read_from_replicas
3573 self.load_balancing_strategy = load_balancing_strategy
3574 self.command_flags = self.__class__.COMMAND_FLAGS.copy()
3575 self.cluster_response_callbacks = cluster_response_callbacks
3576 self.reinitialize_counter = 0
3577 self.reinitialize_steps = reinitialize_steps
3578 if retry is not None:
3579 self.retry = retry
3580 else:
3581 self.retry = Retry(
3582 backoff=ExponentialWithJitterBackoff(
3583 base=DEFAULT_RETRY_BASE, cap=DEFAULT_RETRY_CAP
3584 ),
3585 retries=cluster_error_retry_attempts,
3586 )
3588 self.encoder = Encoder(
3589 kwargs.get("encoding", "utf-8"),
3590 kwargs.get("encoding_errors", "strict"),
3591 kwargs.get("decode_responses", False),
3592 )
3593 if lock is None:
3594 lock = threading.RLock()
3595 self._lock = lock
3596 self.parent_execute_command = super().execute_command
3597 self._execution_strategy: ExecutionStrategy = (
3598 PipelineStrategy(self) if not transaction else TransactionStrategy(self)
3599 )
3601 # For backward compatibility, mapping from existing policies to new one
3602 self._command_flags_mapping: dict[str, Union[RequestPolicy, ResponsePolicy]] = {
3603 self.__class__.RANDOM: RequestPolicy.DEFAULT_KEYLESS,
3604 self.__class__.PRIMARIES: RequestPolicy.ALL_SHARDS,
3605 self.__class__.ALL_NODES: RequestPolicy.ALL_NODES,
3606 self.__class__.REPLICAS: RequestPolicy.ALL_REPLICAS,
3607 self.__class__.DEFAULT_NODE: RequestPolicy.DEFAULT_NODE,
3608 SLOT_ID: RequestPolicy.DEFAULT_KEYED,
3609 }
3611 self._policies_callback_mapping: dict[
3612 Union[RequestPolicy, ResponsePolicy], Callable
3613 ] = {
3614 RequestPolicy.DEFAULT_KEYLESS: lambda command_name: [
3615 self.get_random_primary_or_all_nodes(command_name)
3616 ],
3617 RequestPolicy.DEFAULT_KEYED: lambda command,
3618 *args: self.get_nodes_from_slot(command, *args),
3619 RequestPolicy.DEFAULT_NODE: lambda: [self.get_default_node()],
3620 RequestPolicy.ALL_SHARDS: self.get_primaries,
3621 RequestPolicy.ALL_NODES: self.get_nodes,
3622 RequestPolicy.ALL_REPLICAS: self.get_replicas,
3623 RequestPolicy.MULTI_SHARD: lambda *args,
3624 **kwargs: self._split_multi_shard_command(*args, **kwargs),
3625 RequestPolicy.SPECIAL: self.get_special_nodes,
3626 ResponsePolicy.DEFAULT_KEYLESS: lambda res: res,
3627 ResponsePolicy.DEFAULT_KEYED: lambda res: res,
3628 }
3630 self._policy_resolver = policy_resolver
3632 if event_dispatcher is None:
3633 self._event_dispatcher = EventDispatcher()
3634 else:
3635 self._event_dispatcher = event_dispatcher
3637 def __repr__(self):
3638 """ """
3639 return f"{type(self).__name__}"
3641 def __enter__(self):
3642 """ """
3643 return self
3645 def __exit__(self, exc_type, exc_value, traceback):
3646 """ """
3647 self.reset()
3649 def __del__(self):
3650 try:
3651 self.reset()
3652 except Exception:
3653 pass
3655 def __len__(self):
3656 """ """
3657 return len(self._execution_strategy.command_queue)
3659 def __bool__(self):
3660 "Pipeline instances should always evaluate to True on Python 3+"
3661 return True
3663 def execute_command(self, *args, **kwargs):
3664 """
3665 Wrapper function for pipeline_execute_command
3666 """
3667 return self._execution_strategy.execute_command(*args, **kwargs)
3669 def pipeline_execute_command(self, *args, **options):
3670 """
3671 Stage a command to be executed when execute() is next called
3673 Returns the current Pipeline object back so commands can be
3674 chained together, such as:
3676 pipe = pipe.set('foo', 'bar').incr('baz').decr('bang')
3678 At some other point, you can then run: pipe.execute(),
3679 which will execute all commands queued in the pipe.
3680 """
3681 return self._execution_strategy.execute_command(*args, **options)
3683 def annotate_exception(self, exception, number, command):
3684 """
3685 Provides extra context to the exception prior to it being handled
3686 """
3687 self._execution_strategy.annotate_exception(exception, number, command)
3689 def execute(self, raise_on_error: bool = True) -> List[Any]:
3690 """
3691 Execute all the commands in the current pipeline
3692 """
3694 try:
3695 return self._execution_strategy.execute(raise_on_error)
3696 finally:
3697 self.reset()
3699 def reset(self):
3700 """
3701 Reset back to empty pipeline.
3702 """
3703 self._execution_strategy.reset()
3705 def send_cluster_commands(
3706 self, stack, raise_on_error=True, allow_redirections=True
3707 ):
3708 return self._execution_strategy.send_cluster_commands(
3709 stack, raise_on_error=raise_on_error, allow_redirections=allow_redirections
3710 )
3712 def exists(self, *keys):
3713 return self._execution_strategy.exists(*keys)
3715 def eval(self):
3716 """ """
3717 return self._execution_strategy.eval()
3719 def multi(self):
3720 """
3721 Start a transactional block of the pipeline after WATCH commands
3722 are issued. End the transactional block with `execute`.
3723 """
3724 self._execution_strategy.multi()
3726 def load_scripts(self):
3727 """ """
3728 self._execution_strategy.load_scripts()
3730 def discard(self):
3731 """ """
3732 self._execution_strategy.discard()
3734 def watch(self, *names):
3735 """Watches the values at keys ``names``"""
3736 self._execution_strategy.watch(*names)
3738 def unwatch(self):
3739 """Unwatches all previously specified keys"""
3740 self._execution_strategy.unwatch()
3742 def script_load_for_pipeline(self, *args, **kwargs):
3743 self._execution_strategy.script_load_for_pipeline(*args, **kwargs)
3745 def delete(self, *names):
3746 self._execution_strategy.delete(*names)
3748 def unlink(self, *names):
3749 self._execution_strategy.unlink(*names)
3752def block_pipeline_command(name: str) -> Callable[..., Any]:
3753 """
3754 Prints error because some pipelined commands should
3755 be blocked when running in cluster-mode
3756 """
3758 def inner(*args, **kwargs):
3759 raise RedisClusterException(
3760 f"ERROR: Calling pipelined function {name} is blocked "
3761 f"when running redis in cluster mode..."
3762 )
3764 return inner
3767# Blocked pipeline commands
3768PIPELINE_BLOCKED_COMMANDS = (
3769 "BGREWRITEAOF",
3770 "BGSAVE",
3771 "BITOP",
3772 "BRPOPLPUSH",
3773 "CLIENT GETNAME",
3774 "CLIENT KILL",
3775 "CLIENT LIST",
3776 "CLIENT SETNAME",
3777 "CLIENT",
3778 "CONFIG GET",
3779 "CONFIG RESETSTAT",
3780 "CONFIG REWRITE",
3781 "CONFIG SET",
3782 "CONFIG",
3783 "DBSIZE",
3784 "ECHO",
3785 "EVALSHA",
3786 "FLUSHALL",
3787 "FLUSHDB",
3788 "INFO",
3789 "KEYS",
3790 "LASTSAVE",
3791 "MGET",
3792 "MGET NONATOMIC",
3793 "MOVE",
3794 "MSET",
3795 "MSETEX",
3796 "MSET NONATOMIC",
3797 "MSETNX",
3798 "PFCOUNT",
3799 "PFMERGE",
3800 "PING",
3801 "PUBLISH",
3802 "RANDOMKEY",
3803 "READONLY",
3804 "READWRITE",
3805 "RENAME",
3806 "RENAMENX",
3807 "RPOPLPUSH",
3808 "SAVE",
3809 "SCAN",
3810 "SCRIPT EXISTS",
3811 "SCRIPT FLUSH",
3812 "SCRIPT KILL",
3813 "SCRIPT LOAD",
3814 "SCRIPT",
3815 "SDIFF",
3816 "SDIFFSTORE",
3817 "SENTINEL GET MASTER ADDR BY NAME",
3818 "SENTINEL MASTER",
3819 "SENTINEL MASTERS",
3820 "SENTINEL MONITOR",
3821 "SENTINEL REMOVE",
3822 "SENTINEL SENTINELS",
3823 "SENTINEL SET",
3824 "SENTINEL SLAVES",
3825 "SENTINEL",
3826 "SHUTDOWN",
3827 "SINTER",
3828 "SINTERSTORE",
3829 "SLAVEOF",
3830 "SLOWLOG GET",
3831 "SLOWLOG LEN",
3832 "SLOWLOG RESET",
3833 "SLOWLOG",
3834 "SMOVE",
3835 "SORT",
3836 "SUNION",
3837 "SUNIONSTORE",
3838 "TIME",
3839)
3840for command in PIPELINE_BLOCKED_COMMANDS:
3841 command = command.replace(" ", "_").lower()
3843 setattr(ClusterPipeline, command, block_pipeline_command(command))
3846class PipelineCommand:
3847 """ """
3849 def __init__(self, args, options=None, position=None):
3850 self.args = args
3851 if options is None:
3852 options = {}
3853 self.options = options
3854 self.position = position
3855 self.result = None
3856 self.node = None
3857 self.asking = False
3858 self.command_policies: Optional[CommandPolicies] = None
3861class NodeCommands:
3862 """ """
3864 def __init__(
3865 self, parse_response, connection_pool: ConnectionPool, connection: Connection
3866 ):
3867 """ """
3868 self.parse_response = parse_response
3869 self.connection_pool = connection_pool
3870 self.connection = connection
3871 self.commands = []
3873 def append(self, c):
3874 """ """
3875 self.commands.append(c)
3877 def write(self):
3878 """
3879 Code borrowed from Redis so it can be fixed
3880 """
3881 connection = self.connection
3882 commands = self.commands
3884 # We are going to clobber the commands with the write, so go ahead
3885 # and ensure that nothing is sitting there from a previous run.
3886 for c in commands:
3887 c.result = None
3889 # build up all commands into a single request to increase network perf
3890 # send all the commands and catch connection and timeout errors.
3891 try:
3892 connection.send_packed_command(
3893 connection.pack_commands([c.args for c in commands])
3894 )
3895 except (ConnectionError, TimeoutError) as e:
3896 for c in commands:
3897 c.result = e
3899 def read(self):
3900 """ """
3901 connection = self.connection
3902 for c in self.commands:
3903 # if there is a result on this command,
3904 # it means we ran into an exception
3905 # like a connection error. Trying to parse
3906 # a response on a connection that
3907 # is no longer open will result in a
3908 # connection error raised by redis-py.
3909 # but redis-py doesn't check in parse_response
3910 # that the sock object is
3911 # still set and if you try to
3912 # read from a closed connection, it will
3913 # result in an AttributeError because
3914 # it will do a readline() call on None.
3915 # This can have all kinds of nasty side-effects.
3916 # Treating this case as a connection error
3917 # is fine because it will dump
3918 # the connection object back into the
3919 # pool and on the next write, it will
3920 # explicitly open the connection and all will be well.
3921 if c.result is None:
3922 try:
3923 c.result = self.parse_response(connection, c.args[0], **c.options)
3924 except (ConnectionError, TimeoutError) as e:
3925 for c in self.commands:
3926 c.result = e
3927 return
3928 except RedisError:
3929 c.result = sys.exc_info()[1]
3932class ExecutionStrategy(ABC):
3933 @property
3934 @abstractmethod
3935 def command_queue(self):
3936 pass
3938 @abstractmethod
3939 def execute_command(self, *args, **kwargs):
3940 """
3941 Execution flow for current execution strategy.
3943 See: ClusterPipeline.execute_command()
3944 """
3945 pass
3947 @abstractmethod
3948 def annotate_exception(self, exception, number, command):
3949 """
3950 Annotate exception according to current execution strategy.
3952 See: ClusterPipeline.annotate_exception()
3953 """
3954 pass
3956 @abstractmethod
3957 def pipeline_execute_command(self, *args, **options):
3958 """
3959 Pipeline execution flow for current execution strategy.
3961 See: ClusterPipeline.pipeline_execute_command()
3962 """
3963 pass
3965 @abstractmethod
3966 def execute(self, raise_on_error: bool = True) -> List[Any]:
3967 """
3968 Executes current execution strategy.
3970 See: ClusterPipeline.execute()
3971 """
3972 pass
3974 @abstractmethod
3975 def send_cluster_commands(
3976 self, stack, raise_on_error=True, allow_redirections=True
3977 ):
3978 """
3979 Sends commands according to current execution strategy.
3981 See: ClusterPipeline.send_cluster_commands()
3982 """
3983 pass
3985 @abstractmethod
3986 def reset(self):
3987 """
3988 Resets current execution strategy.
3990 See: ClusterPipeline.reset()
3991 """
3992 pass
3994 @abstractmethod
3995 def exists(self, *keys):
3996 pass
3998 @abstractmethod
3999 def eval(self):
4000 pass
4002 @abstractmethod
4003 def multi(self):
4004 """
4005 Starts transactional context.
4007 See: ClusterPipeline.multi()
4008 """
4009 pass
4011 @abstractmethod
4012 def load_scripts(self):
4013 pass
4015 @abstractmethod
4016 def watch(self, *names):
4017 pass
4019 @abstractmethod
4020 def unwatch(self):
4021 """
4022 Unwatches all previously specified keys
4024 See: ClusterPipeline.unwatch()
4025 """
4026 pass
4028 @abstractmethod
4029 def script_load_for_pipeline(self, *args, **kwargs):
4030 pass
4032 @abstractmethod
4033 def delete(self, *names):
4034 """
4035 "Delete a key specified by ``names``"
4037 See: ClusterPipeline.delete()
4038 """
4039 pass
4041 @abstractmethod
4042 def unlink(self, *names):
4043 """
4044 "Unlink a key specified by ``names``"
4046 See: ClusterPipeline.unlink()
4047 """
4048 pass
4050 @abstractmethod
4051 def discard(self):
4052 pass
4055class AbstractStrategy(ExecutionStrategy):
4056 def __init__(
4057 self,
4058 pipe: ClusterPipeline,
4059 ):
4060 self._command_queue: List[PipelineCommand] = []
4061 self._pipe = pipe
4062 self._nodes_manager = self._pipe.nodes_manager
4064 @property
4065 def command_queue(self):
4066 return self._command_queue
4068 @command_queue.setter
4069 def command_queue(self, queue: List[PipelineCommand]):
4070 self._command_queue = queue
4072 @abstractmethod
4073 def execute_command(self, *args, **kwargs):
4074 pass
4076 def pipeline_execute_command(self, *args, **options):
4077 self._command_queue.append(
4078 PipelineCommand(args, options, len(self._command_queue))
4079 )
4080 return self._pipe
4082 def _himport_prepare_pipeline(self, redis_node, conn, commands):
4083 """Delegate to the shared sync HIMPORT executor."""
4084 _himport_exec.prepare_pipeline(redis_node, conn, [args for args, _ in commands])
4086 @abstractmethod
4087 def execute(self, raise_on_error: bool = True) -> List[Any]:
4088 pass
4090 @abstractmethod
4091 def send_cluster_commands(
4092 self, stack, raise_on_error=True, allow_redirections=True
4093 ):
4094 pass
4096 @abstractmethod
4097 def reset(self):
4098 pass
4100 def exists(self, *keys):
4101 return self.execute_command("EXISTS", *keys)
4103 def eval(self):
4104 """ """
4105 raise RedisClusterException("method eval() is not implemented")
4107 def load_scripts(self):
4108 """ """
4109 raise RedisClusterException("method load_scripts() is not implemented")
4111 def script_load_for_pipeline(self, *args, **kwargs):
4112 """ """
4113 raise RedisClusterException(
4114 "method script_load_for_pipeline() is not implemented"
4115 )
4117 def annotate_exception(self, exception, number, command):
4118 """
4119 Provides extra context to the exception prior to it being handled
4120 """
4121 cmd = " ".join(map(safe_str, command))
4122 msg = (
4123 f"Command # {number} ({truncate_text(cmd)}) of pipeline "
4124 f"caused error: {exception.args[0]}"
4125 )
4126 exception.args = (msg,) + exception.args[1:]
4129class PipelineStrategy(AbstractStrategy):
4130 def __init__(self, pipe: ClusterPipeline):
4131 super().__init__(pipe)
4132 self.command_flags = pipe.command_flags
4134 def execute_command(self, *args, **kwargs):
4135 return self.pipeline_execute_command(*args, **kwargs)
4137 def _raise_first_error(self, stack, start_time):
4138 """
4139 Raise the first exception on the stack
4140 """
4141 for c in stack:
4142 r = c.result
4143 if isinstance(r, Exception):
4144 self.annotate_exception(r, c.position + 1, c.args)
4146 record_operation_duration(
4147 command_name="PIPELINE",
4148 duration_seconds=time.monotonic() - start_time,
4149 error=r,
4150 )
4152 raise r
4154 def execute(self, raise_on_error: bool = True) -> List[Any]:
4155 stack = self._command_queue
4156 if not stack:
4157 return []
4159 try:
4160 return self.send_cluster_commands(stack, raise_on_error)
4161 finally:
4162 self.reset()
4164 def reset(self):
4165 """
4166 Reset back to empty pipeline.
4167 """
4168 self._command_queue = []
4170 def send_cluster_commands(
4171 self, stack, raise_on_error=True, allow_redirections=True
4172 ):
4173 """
4174 Wrapper for RedisCluster.ERRORS_ALLOW_RETRY errors handling.
4176 If one of the retryable exceptions has been thrown we assume that:
4177 - connection_pool was disconnected
4178 - connection_pool was reset
4179 - refresh_table_asap set to True
4181 It will try the number of times specified by
4182 the retries in config option "self.retry"
4183 which defaults to 10 unless manually configured.
4185 If it reaches the number of times, the command will
4186 raises ClusterDownException.
4187 """
4188 if not stack:
4189 return []
4190 retry_attempts = self._pipe.retry.get_retries()
4191 while True:
4192 try:
4193 return self._send_cluster_commands(
4194 stack,
4195 raise_on_error=raise_on_error,
4196 allow_redirections=allow_redirections,
4197 )
4198 except RedisCluster.ERRORS_ALLOW_RETRY as e:
4199 if retry_attempts > 0:
4200 # Try again with the new cluster setup. All other errors
4201 # should be raised.
4202 retry_attempts -= 1
4203 pass
4204 else:
4205 raise e
4207 def _send_cluster_commands(
4208 self, stack, raise_on_error=True, allow_redirections=True
4209 ):
4210 """
4211 Send a bunch of cluster commands to the redis cluster.
4213 `allow_redirections` If the pipeline should follow
4214 `ASK` & `MOVED` responses automatically. If set
4215 to false it will raise RedisClusterException.
4216 """
4217 # the first time sending the commands we send all of
4218 # the commands that were queued up.
4219 # if we have to run through it again, we only retry
4220 # the commands that failed.
4221 attempt = sorted(stack, key=lambda x: x.position)
4222 is_default_node = False
4223 # build a list of node objects based on node names we need to
4224 nodes: dict[str, NodeCommands] = {}
4225 # node objects keyed by name, so each node's connection can be pre-flighted
4226 # for HIMPORT SET (PREPARE) before the batched write.
4227 node_objs: dict = {}
4228 nodes_written = 0
4229 nodes_read = 0
4231 try:
4232 # as we move through each command that still needs to be processed,
4233 # we figure out the slot number that command maps to, then from
4234 # the slot determine the node.
4235 for c in attempt:
4236 command_policies = self._pipe._policy_resolver.resolve(
4237 c.args[0].lower()
4238 )
4239 # refer to our internal node -> slot table that
4240 # tells us where a given command should route to.
4241 # (it might be possible we have a cached node that no longer
4242 # exists in the cluster, which is why we do this in a loop)
4243 passed_targets = c.options.pop("target_nodes", None)
4244 if passed_targets and not self._is_nodes_flag(passed_targets):
4245 target_nodes = self._parse_target_nodes(passed_targets)
4247 if not command_policies:
4248 command_policies = CommandPolicies()
4249 else:
4250 if not command_policies:
4251 command = c.args[0].upper()
4252 if (
4253 len(c.args) >= 2
4254 and f"{c.args[0]} {c.args[1]}".upper()
4255 in self._pipe.command_flags
4256 ):
4257 command = f"{c.args[0]} {c.args[1]}".upper()
4259 # We only could resolve key properties if command is not
4260 # in a list of pre-defined request policies
4261 command_flag = self.command_flags.get(command)
4262 if not command_flag:
4263 # Fallback to default policy
4264 if not self._pipe.get_default_node():
4265 keys = None
4266 else:
4267 keys = self._pipe._get_command_keys(*c.args)
4268 if not keys or len(keys) == 0:
4269 command_policies = CommandPolicies()
4270 else:
4271 command_policies = CommandPolicies(
4272 request_policy=RequestPolicy.DEFAULT_KEYED,
4273 response_policy=ResponsePolicy.DEFAULT_KEYED,
4274 )
4275 else:
4276 if command_flag in self._pipe._command_flags_mapping:
4277 command_policies = CommandPolicies(
4278 request_policy=self._pipe._command_flags_mapping[
4279 command_flag
4280 ]
4281 )
4282 else:
4283 command_policies = CommandPolicies()
4285 target_nodes = self._determine_nodes(
4286 *c.args,
4287 request_policy=command_policies.request_policy,
4288 node_flag=passed_targets,
4289 )
4290 if not target_nodes:
4291 raise RedisClusterException(
4292 f"No targets were found to execute {c.args} command on"
4293 )
4294 c.command_policies = command_policies
4295 if len(target_nodes) > 1:
4296 raise RedisClusterException(
4297 f"Too many targets for command {c.args}"
4298 )
4300 node = target_nodes[0]
4301 if node == self._pipe.get_default_node():
4302 is_default_node = True
4304 # now that we know the name of the node
4305 # ( it's just a string in the form of host:port )
4306 # we can build a list of commands for each node.
4307 node_name = node.name
4308 if node_name not in nodes:
4309 redis_node = self._pipe.get_redis_connection(node)
4310 try:
4311 connection = get_connection(redis_node)
4312 except (ConnectionError, TimeoutError):
4313 # Release any connections we've already acquired before clearing nodes
4314 for n in nodes.values():
4315 n.connection_pool.release(n.connection)
4316 # Connection retries are being handled in the node's
4317 # Retry object. Reinitialize the node -> slot table.
4318 self._nodes_manager.initialize()
4319 if is_default_node:
4320 self._pipe.replace_default_node()
4321 nodes = {}
4322 raise
4323 nodes[node_name] = NodeCommands(
4324 redis_node.parse_response,
4325 redis_node.connection_pool,
4326 connection,
4327 )
4328 node_objs[node_name] = node
4329 nodes[node_name].append(c)
4331 # send the commands in sequence.
4332 # we write to all the open sockets for each node first,
4333 # before reading anything
4334 # this allows us to flush all the requests out across the
4335 # network
4336 # so that we can read them from different sockets as they come back.
4337 # we don't multiplex on the sockets as they come available,
4338 # but that shouldn't make too much difference.
4340 # HIMPORT SETs in the batch need their fieldsets prepared on each
4341 # node's connection first; the packed write bypasses the per-command
4342 # lazy prepare, so pre-flight the PREPARE (once per node) here.
4343 for node_name, n in nodes.items():
4344 redis_node = self._pipe.get_redis_connection(node_objs[node_name])
4345 self._himport_prepare_pipeline(
4346 redis_node, n.connection, [(c.args, c.options) for c in n.commands]
4347 )
4349 # Start timing for observability
4350 start_time = time.monotonic()
4352 node_commands = nodes.values()
4353 for n in node_commands:
4354 nodes_written += 1
4355 n.write()
4357 for n in node_commands:
4358 n.read()
4360 # Find the first error in this node's commands, if any
4361 node_error = None
4362 for cmd in n.commands:
4363 if isinstance(cmd.result, Exception):
4364 node_error = cmd.result
4365 break
4367 record_operation_duration(
4368 command_name="PIPELINE",
4369 duration_seconds=time.monotonic() - start_time,
4370 server_address=n.connection.host,
4371 server_port=n.connection.port,
4372 db_namespace=str(n.connection.db),
4373 error=node_error,
4374 )
4375 nodes_read += 1
4376 finally:
4377 # release all the redis connections we allocated earlier
4378 # back into the connection pool.
4379 # if the connection is dirty (that is: we've written
4380 # commands to it, but haven't read the responses), we need
4381 # to close the connection before returning it to the pool.
4382 # otherwise, the next caller to use this connection will
4383 # read the response from _this_ request, not its own request.
4384 # disconnecting discards the dirty state & forces the next
4385 # caller to reconnect.
4386 # NOTE: dicts have a consistent ordering; we're iterating
4387 # through nodes.values() in the same order as we are when
4388 # reading / writing to the connections above, which is critical
4389 # for how we're using the nodes_written/nodes_read offsets.
4390 for i, n in enumerate(nodes.values()):
4391 if i < nodes_written and i >= nodes_read:
4392 n.connection.disconnect()
4393 n.connection_pool.release(n.connection)
4395 # if the response isn't an exception it is a
4396 # valid response from the node
4397 # we're all done with that command, YAY!
4398 # if we have more commands to attempt, we've run into problems.
4399 # collect all the commands we are allowed to retry.
4400 # (MOVED, ASK, or connection errors or timeout errors)
4401 attempt = sorted(
4402 (
4403 c
4404 for c in attempt
4405 if isinstance(c.result, ClusterPipeline.ERRORS_ALLOW_RETRY)
4406 ),
4407 key=lambda x: x.position,
4408 )
4409 if attempt and allow_redirections:
4410 # RETRY MAGIC HAPPENS HERE!
4411 # send these remaining commands one at a time using `execute_command`
4412 # in the main client. This keeps our retry logic
4413 # in one place mostly,
4414 # and allows us to be more confident in correctness of behavior.
4415 # at this point any speed gains from pipelining have been lost
4416 # anyway, so we might as well make the best
4417 # attempt to get the correct behavior.
4418 #
4419 # The client command will handle retries for each
4420 # individual command sequentially as we pass each
4421 # one into `execute_command`. Any exceptions
4422 # that bubble out should only appear once all
4423 # retries have been exhausted.
4424 #
4425 # If a lot of commands have failed, we'll be setting the
4426 # flag to rebuild the slots table from scratch.
4427 # So MOVED errors should correct themselves fairly quickly.
4428 self._pipe.reinitialize_counter += 1
4429 if self._pipe._should_reinitialized():
4430 self._nodes_manager.initialize()
4431 if is_default_node:
4432 self._pipe.replace_default_node()
4433 for c in attempt:
4434 try:
4435 # send each command individually like we
4436 # do in the main client.
4437 c.result = self._pipe.parent_execute_command(*c.args, **c.options)
4438 except RedisError as e:
4439 c.result = e
4441 # turn the response back into a simple flat array that corresponds
4442 # to the sequence of commands issued in the stack in pipeline.execute()
4443 response = []
4444 for c in sorted(stack, key=lambda x: x.position):
4445 if c.args[0] in self._pipe.cluster_response_callbacks:
4446 # Remove keys entry, it needs only for cache.
4447 c.options.pop("keys", None)
4448 c.result = self._pipe._policies_callback_mapping[
4449 c.command_policies.response_policy
4450 ](
4451 self._pipe.cluster_response_callbacks[c.args[0]](
4452 c.result, **c.options
4453 )
4454 )
4455 response.append(c.result)
4457 if raise_on_error:
4458 self._raise_first_error(stack, start_time)
4460 return response
4462 def _is_nodes_flag(self, target_nodes):
4463 return isinstance(target_nodes, str) and target_nodes in self._pipe.node_flags
4465 def _parse_target_nodes(self, target_nodes):
4466 if isinstance(target_nodes, list):
4467 nodes = target_nodes
4468 elif isinstance(target_nodes, ClusterNode):
4469 # Supports passing a single ClusterNode as a variable
4470 nodes = [target_nodes]
4471 elif isinstance(target_nodes, dict):
4472 # Supports dictionaries of the format {node_name: node}.
4473 # It enables to execute commands with multi nodes as follows:
4474 # rc.cluster_save_config(rc.get_primaries())
4475 nodes = target_nodes.values()
4476 else:
4477 raise TypeError(
4478 "target_nodes type can be one of the following: "
4479 "node_flag (PRIMARIES, REPLICAS, RANDOM, ALL_NODES),"
4480 "ClusterNode, list<ClusterNode>, or dict<any, ClusterNode>. "
4481 f"The passed type is {type(target_nodes)}"
4482 )
4483 return nodes
4485 def _determine_nodes(
4486 self, *args, request_policy: RequestPolicy, **kwargs
4487 ) -> List["ClusterNode"]:
4488 # Determine which nodes should be executed the command on.
4489 # Returns a list of target nodes.
4490 command = args[0].upper()
4491 if (
4492 len(args) >= 2
4493 and f"{args[0]} {args[1]}".upper() in self._pipe.command_flags
4494 ):
4495 command = f"{args[0]} {args[1]}".upper()
4497 nodes_flag = kwargs.pop("nodes_flag", None)
4498 if nodes_flag is not None:
4499 # nodes flag passed by the user
4500 command_flag = nodes_flag
4501 else:
4502 # get the nodes group for this command if it was predefined
4503 command_flag = self._pipe.command_flags.get(command)
4505 if command_flag in self._pipe._command_flags_mapping:
4506 request_policy = self._pipe._command_flags_mapping[command_flag]
4508 policy_callback = self._pipe._policies_callback_mapping[request_policy]
4510 if request_policy == RequestPolicy.DEFAULT_KEYED:
4511 nodes = policy_callback(command, *args)
4512 elif request_policy == RequestPolicy.MULTI_SHARD:
4513 nodes = policy_callback(*args, **kwargs)
4514 elif request_policy == RequestPolicy.DEFAULT_KEYLESS:
4515 nodes = policy_callback(args[0])
4516 else:
4517 nodes = policy_callback()
4519 if args[0].lower() == "ft.aggregate":
4520 self._aggregate_nodes = nodes
4522 return nodes
4524 def multi(self):
4525 raise RedisClusterException(
4526 "method multi() is not supported outside of transactional context"
4527 )
4529 def discard(self):
4530 raise RedisClusterException(
4531 "method discard() is not supported outside of transactional context"
4532 )
4534 def watch(self, *names):
4535 raise RedisClusterException(
4536 "method watch() is not supported outside of transactional context"
4537 )
4539 def unwatch(self, *names):
4540 raise RedisClusterException(
4541 "method unwatch() is not supported outside of transactional context"
4542 )
4544 def delete(self, *names):
4545 if len(names) != 1:
4546 raise RedisClusterException(
4547 "deleting multiple keys is not implemented in pipeline command"
4548 )
4550 return self.execute_command("DEL", names[0])
4552 def unlink(self, *names):
4553 if len(names) != 1:
4554 raise RedisClusterException(
4555 "unlinking multiple keys is not implemented in pipeline command"
4556 )
4558 return self.execute_command("UNLINK", names[0])
4561class TransactionStrategy(AbstractStrategy):
4562 NO_SLOTS_COMMANDS = {"UNWATCH"}
4563 IMMEDIATE_EXECUTE_COMMANDS = {"WATCH", "UNWATCH"}
4564 UNWATCH_COMMANDS = {"DISCARD", "EXEC", "UNWATCH"}
4565 SLOT_REDIRECT_ERRORS = (AskError, MovedError)
4566 CONNECTION_ERRORS = (
4567 ConnectionError,
4568 OSError,
4569 ClusterDownError,
4570 SlotNotCoveredError,
4571 )
4573 def __init__(self, pipe: ClusterPipeline):
4574 super().__init__(pipe)
4575 self._explicit_transaction = False
4576 self._watching = False
4577 self._pipeline_slots: Set[int] = set()
4578 self._transaction_connection: Optional[Connection] = None
4579 self._executing = False
4580 self._retry = copy(self._pipe.retry)
4581 self._retry.update_supported_errors(
4582 RedisCluster.ERRORS_ALLOW_RETRY + self.SLOT_REDIRECT_ERRORS
4583 )
4585 def _get_client_and_connection_for_transaction(self) -> Tuple[Redis, Connection]:
4586 """
4587 Find a connection for a pipeline transaction.
4589 For running an atomic transaction, watch keys ensure that contents have not been
4590 altered as long as the watch commands for those keys were sent over the same
4591 connection. So once we start watching a key, we fetch a connection to the
4592 node that owns that slot and reuse it.
4593 """
4594 if not self._pipeline_slots:
4595 raise RedisClusterException(
4596 "At least a command with a key is needed to identify a node"
4597 )
4599 node: ClusterNode = self._nodes_manager.get_node_from_slot(
4600 list(self._pipeline_slots)[0], False
4601 )
4602 redis_node: Redis = self._pipe.get_redis_connection(node)
4603 if self._transaction_connection:
4604 if not redis_node.connection_pool.owns_connection(
4605 self._transaction_connection
4606 ):
4607 previous_node = self._nodes_manager.find_connection_owner(
4608 self._transaction_connection
4609 )
4610 previous_node.connection_pool.release(self._transaction_connection)
4611 self._transaction_connection = None
4613 if not self._transaction_connection:
4614 self._transaction_connection = get_connection(redis_node)
4616 return redis_node, self._transaction_connection
4618 def execute_command(self, *args, **kwargs):
4619 slot_number: Optional[int] = None
4620 if args[0] not in ClusterPipeline.NO_SLOTS_COMMANDS:
4621 slot_number = self._pipe.determine_slot(*args)
4623 if (
4624 self._watching or args[0] in self.IMMEDIATE_EXECUTE_COMMANDS
4625 ) and not self._explicit_transaction:
4626 if args[0] == "WATCH":
4627 self._validate_watch()
4629 if slot_number is not None:
4630 if self._pipeline_slots and slot_number not in self._pipeline_slots:
4631 raise CrossSlotTransactionError(
4632 "Cannot watch or send commands on different slots"
4633 )
4635 self._pipeline_slots.add(slot_number)
4636 elif args[0] not in self.NO_SLOTS_COMMANDS:
4637 raise RedisClusterException(
4638 f"Cannot identify slot number for command: {args[0]},"
4639 "it cannot be triggered in a transaction"
4640 )
4642 return self._immediate_execute_command(*args, **kwargs)
4643 else:
4644 if slot_number is not None:
4645 self._pipeline_slots.add(slot_number)
4647 return self.pipeline_execute_command(*args, **kwargs)
4649 def _validate_watch(self):
4650 if self._explicit_transaction:
4651 raise RedisError("Cannot issue a WATCH after a MULTI")
4653 self._watching = True
4655 def _immediate_execute_command(self, *args, **options):
4656 return self._retry.call_with_retry(
4657 lambda: self._get_connection_and_send_command(*args, **options),
4658 self._reinitialize_on_error,
4659 with_failure_count=True,
4660 )
4662 def _get_connection_and_send_command(self, *args, **options):
4663 redis_node, connection = self._get_client_and_connection_for_transaction()
4665 # Start timing for observability
4666 start_time = time.monotonic()
4668 try:
4669 response = self._send_command_parse_response(
4670 connection, redis_node, args[0], *args, **options
4671 )
4673 record_operation_duration(
4674 command_name=args[0],
4675 duration_seconds=time.monotonic() - start_time,
4676 server_address=connection.host,
4677 server_port=connection.port,
4678 db_namespace=str(connection.db),
4679 )
4681 return response
4682 except Exception as e:
4683 if connection:
4684 # this is used to report the metrics based on host and port info
4685 e.connection = connection
4686 record_operation_duration(
4687 command_name=args[0],
4688 duration_seconds=time.monotonic() - start_time,
4689 server_address=connection.host,
4690 server_port=connection.port,
4691 db_namespace=str(connection.db),
4692 error=e,
4693 )
4694 raise
4696 def _send_command_parse_response(
4697 self, conn, redis_node: Redis, command_name, *args, **options
4698 ):
4699 """
4700 Send a command and parse the response
4701 """
4703 # HIMPORT SET's wire form depends on per-connection state: the fieldset
4704 # must be PREPAREd on this connection first, and any fieldset discarded
4705 # since this connection last reconciled must be dropped. The
4706 # immediate/watched path (commands issued after WATCH, before MULTI)
4707 # would otherwise send a bare HIMPORT SET and fail with "no such
4708 # fieldset". Route it through the node's HIMPORT executor, the same way
4709 # the normal cluster path, the batched MULTI/EXEC path, and standalone
4710 # watched pipelines all do.
4711 himport_set = parse_himport_set_args(args)
4712 if himport_set is not None:
4713 # HIMPORT SET in the joined or split raw form; operands at the right
4714 # offsets. Too few operands returns None and falls through to the bare
4715 # send so the server returns its arity error.
4716 key, fieldset_name, values = himport_set
4717 output = redis_node._himport_execute_set(conn, key, fieldset_name, values)
4718 else:
4719 conn.send_command(*args)
4720 output = redis_node.parse_response(conn, command_name, **options)
4722 if command_name in self.UNWATCH_COMMANDS:
4723 self._watching = False
4724 return output
4726 def _reinitialize_on_error(self, error, failure_count):
4727 if hasattr(error, "connection"):
4728 record_error_count(
4729 server_address=error.connection.host,
4730 server_port=error.connection.port,
4731 network_peer_address=error.connection.host,
4732 network_peer_port=error.connection.port,
4733 error_type=error,
4734 retry_attempts=failure_count,
4735 is_internal=True,
4736 )
4738 if self._watching:
4739 if type(error) in self.SLOT_REDIRECT_ERRORS and self._executing:
4740 raise WatchError("Slot rebalancing occurred while watching keys")
4742 if (
4743 type(error) in self.SLOT_REDIRECT_ERRORS
4744 or type(error) in self.CONNECTION_ERRORS
4745 ):
4746 if self._transaction_connection:
4747 if is_debug_log_enabled():
4748 logger.debug(
4749 f"Operation failed, "
4750 f"with connection: {self._transaction_connection}, "
4751 f"details: {self._transaction_connection.extract_connection_details()}",
4752 )
4753 # Disconnect and release back to pool
4754 self._transaction_connection.disconnect()
4755 node = self._nodes_manager.find_connection_owner(
4756 self._transaction_connection
4757 )
4758 if node and node.redis_connection:
4759 node.redis_connection.connection_pool.release(
4760 self._transaction_connection
4761 )
4762 self._transaction_connection = None
4764 self._pipe.reinitialize_counter += 1
4765 if self._pipe._should_reinitialized():
4766 self._nodes_manager.initialize()
4767 self.reinitialize_counter = 0
4768 else:
4769 if isinstance(error, AskError):
4770 self._nodes_manager.move_slot(error)
4772 self._executing = False
4774 def _raise_first_error(self, responses, stack, start_time):
4775 """
4776 Raise the first exception on the stack
4777 """
4778 for r, cmd in zip(responses, stack):
4779 if isinstance(r, Exception):
4780 self.annotate_exception(r, cmd.position + 1, cmd.args)
4782 record_operation_duration(
4783 command_name="TRANSACTION",
4784 duration_seconds=time.monotonic() - start_time,
4785 server_address=self._transaction_connection.host,
4786 server_port=self._transaction_connection.port,
4787 db_namespace=str(self._transaction_connection.db),
4788 )
4790 raise r
4792 def execute(self, raise_on_error: bool = True) -> List[Any]:
4793 stack = self._command_queue
4794 if not stack and (not self._watching or not self._pipeline_slots):
4795 return []
4797 return self._execute_transaction_with_retries(stack, raise_on_error)
4799 def _execute_transaction_with_retries(
4800 self, stack: List["PipelineCommand"], raise_on_error: bool
4801 ):
4802 return self._retry.call_with_retry(
4803 lambda: self._execute_transaction(stack, raise_on_error),
4804 lambda error, failure_count: self._reinitialize_on_error(
4805 error, failure_count
4806 ),
4807 with_failure_count=True,
4808 )
4810 def _execute_transaction(
4811 self, stack: List["PipelineCommand"], raise_on_error: bool
4812 ):
4813 if len(self._pipeline_slots) > 1:
4814 raise CrossSlotTransactionError(
4815 "All keys involved in a cluster transaction must map to the same slot"
4816 )
4818 self._executing = True
4820 redis_node, connection = self._get_client_and_connection_for_transaction()
4822 # Ensure fieldsets referenced by buffered HIMPORT SETs are prepared on this
4823 # node's connection before the MULTI/EXEC block (session state, not
4824 # transactional). All keys share one slot here, so it is a single node.
4825 self._himport_prepare_pipeline(
4826 redis_node, connection, [(c.args, c.options) for c in stack]
4827 )
4829 stack = chain(
4830 [PipelineCommand(("MULTI",))],
4831 stack,
4832 [PipelineCommand(("EXEC",))],
4833 )
4834 commands = [c.args for c in stack if EMPTY_RESPONSE not in c.options]
4835 packed_commands = connection.pack_commands(commands)
4837 # Start timing for observability
4838 start_time = time.monotonic()
4840 connection.send_packed_command(packed_commands)
4841 errors = []
4843 # parse off the response for MULTI
4844 # NOTE: we need to handle ResponseErrors here and continue
4845 # so that we read all the additional command messages from
4846 # the socket
4847 try:
4848 redis_node.parse_response(connection, "MULTI")
4849 except ResponseError as e:
4850 self.annotate_exception(e, 0, "MULTI")
4851 errors.append(e)
4852 except self.CONNECTION_ERRORS as cluster_error:
4853 self.annotate_exception(cluster_error, 0, "MULTI")
4854 raise
4856 # and all the other commands
4857 for i, command in enumerate(self._command_queue):
4858 if EMPTY_RESPONSE in command.options:
4859 errors.append((i, command.options[EMPTY_RESPONSE]))
4860 else:
4861 try:
4862 _ = redis_node.parse_response(connection, "_")
4863 except self.SLOT_REDIRECT_ERRORS as slot_error:
4864 self.annotate_exception(slot_error, i + 1, command.args)
4865 errors.append(slot_error)
4866 except self.CONNECTION_ERRORS as cluster_error:
4867 self.annotate_exception(cluster_error, i + 1, command.args)
4868 raise
4869 except ResponseError as e:
4870 self.annotate_exception(e, i + 1, command.args)
4871 errors.append(e)
4873 response = None
4874 # parse the EXEC.
4875 try:
4876 response = redis_node.parse_response(connection, "EXEC")
4877 except ExecAbortError:
4878 if errors:
4879 raise errors[0]
4880 raise
4882 self._executing = False
4884 record_operation_duration(
4885 command_name="TRANSACTION",
4886 duration_seconds=time.monotonic() - start_time,
4887 server_address=connection.host,
4888 server_port=connection.port,
4889 db_namespace=str(connection.db),
4890 )
4892 # EXEC clears any watched keys
4893 self._watching = False
4895 if response is None:
4896 raise WatchError("Watched variable changed.")
4898 # put any parse errors into the response
4899 for i, e in errors:
4900 response.insert(i, e)
4902 if len(response) != len(self._command_queue):
4903 raise InvalidPipelineStack(
4904 "Unexpected response length for cluster pipeline EXEC."
4905 " Command stack was {} but response had length {}".format(
4906 [c.args[0] for c in self._command_queue], len(response)
4907 )
4908 )
4910 # find any errors in the response and raise if necessary
4911 if raise_on_error or len(errors) > 0:
4912 self._raise_first_error(
4913 response,
4914 self._command_queue,
4915 start_time,
4916 )
4918 # We have to run response callbacks manually
4919 data = []
4920 for r, cmd in zip(response, self._command_queue):
4921 if not isinstance(r, Exception):
4922 command_name = cmd.args[0]
4923 if command_name in self._pipe.cluster_response_callbacks:
4924 r = self._pipe.cluster_response_callbacks[command_name](
4925 r, **cmd.options
4926 )
4927 data.append(r)
4928 return data
4930 def reset(self):
4931 self._command_queue = []
4933 # make sure to reset the connection state in the event that we were
4934 # watching something
4935 if self._transaction_connection:
4936 try:
4937 if self._watching:
4938 # call this manually since our unwatch or
4939 # immediate_execute_command methods can call reset()
4940 self._transaction_connection.send_command("UNWATCH")
4941 self._transaction_connection.read_response()
4942 # we can safely return the connection to the pool here since we're
4943 # sure we're no longer WATCHing anything
4944 node = self._nodes_manager.find_connection_owner(
4945 self._transaction_connection
4946 )
4947 if node and node.redis_connection:
4948 node.redis_connection.connection_pool.release(
4949 self._transaction_connection
4950 )
4951 self._transaction_connection = None
4952 except self.CONNECTION_ERRORS:
4953 # disconnect will also remove any previous WATCHes
4954 if self._transaction_connection:
4955 self._transaction_connection.disconnect()
4956 node = self._nodes_manager.find_connection_owner(
4957 self._transaction_connection
4958 )
4959 if node and node.redis_connection:
4960 node.redis_connection.connection_pool.release(
4961 self._transaction_connection
4962 )
4963 self._transaction_connection = None
4965 # clean up the other instance attributes
4966 self._watching = False
4967 self._explicit_transaction = False
4968 self._pipeline_slots = set()
4969 self._executing = False
4971 def send_cluster_commands(
4972 self, stack, raise_on_error=True, allow_redirections=True
4973 ):
4974 raise NotImplementedError(
4975 "send_cluster_commands cannot be executed in transactional context."
4976 )
4978 def multi(self):
4979 if self._explicit_transaction:
4980 raise RedisError("Cannot issue nested calls to MULTI")
4981 if self._command_queue:
4982 raise RedisError(
4983 "Commands without an initial WATCH have already been issued"
4984 )
4985 self._explicit_transaction = True
4987 def watch(self, *names):
4988 if self._explicit_transaction:
4989 raise RedisError("Cannot issue a WATCH after a MULTI")
4991 return self.execute_command("WATCH", *names)
4993 def unwatch(self):
4994 if self._watching:
4995 return self.execute_command("UNWATCH")
4997 return True
4999 def discard(self):
5000 self.reset()
5002 def delete(self, *names):
5003 return self.execute_command("DEL", *names)
5005 def unlink(self, *names):
5006 return self.execute_command("UNLINK", *names)