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