1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

from redis.asyncio.cluster import RedisCluster as AsyncRedisCluster

...

REDIS_CONNECTION_PARAMS = {
	"password": app_settings.redis_password,
	"username": app_settings.redis_user,
	"host": app_settings.redis_endpoint,
	"port": app_settings.redis_port,
	"encoding": "utf-8",
	"ssl": True,
	"decode_responses": True,
	"require_full_coverage": False,
}

self._redis_cluster = AsyncRedisCluster(**REDIS_CONNECTION_PARAMS)
...

1

Redis Cluster cannot be connected. Please provide at least one reachable node: None

1
2
3
4
5
6
7
8
9

# Linux, macOS, Unix
aws memorydb failover-shard \  
   --cluster-name my-cluster \  
   --shard-name 0001

# Windows
aws memorydb failover-shard ^  
   --cluster-name my-cluster ^  
   --shard-name 0001

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

{  
     "Events": [  
        {  
             "SourceName": "my-cluster",  
             "SourceType": "cluster",  
             "Message": "Failover to replica node my-cluster-0001-002 completed",  
             "Date": "2021-08-22T12:39:37.568000-07:00"  
         },  
         {  
             "SourceName": "my-cluster",  
             "SourceType": "cluster",  
             "Message": "Starting failover for shard 0001",  
             "Date": "2021-08-22T12:39:10.173000-07:00"  
         }  
    ]  
 }

1
2
3
4
5
6
7
8

class MovedError(AskError):
    """
    Error indicated MOVED error received from cluster.
    A request sent to a node that doesn't serve this key will be replayed with
    a MOVED error that points to the correct node.
    """

    pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

...
	except MovedError as e:
		# First, we will try to patch the slots/nodes cache with the
		# redirected node output and try again. If MovedError exceeds
		# 'reinitialize_steps' number of times, we will force
		# reinitializing the tables, and then try again.
		# 'reinitialize_steps' counter will increase faster when
		# the same client object is shared between multiple threads. To
		# reduce the frequency you can set this variable in the
		# RedisCluster constructor.
		self.reinitialize_counter += 1
		if self._should_reinitialized():
			self.nodes_manager.initialize()
			# Reset the counter
			self.reinitialize_counter = 0
		else:
			self.nodes_manager.update_moved_exception(e)

		moved = True
...

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

def _should_reinitialized(self):
	# To reinitialize the cluster on every MOVED error,
	# set reinitialize_steps to 1.
	# To avoid reinitializing the cluster on moved errors, set
	# reinitialize_steps to 0.

	if self.reinitialize_steps  0:
		return False
	else:
		return self.reinitialize_counter % self.reinitialize_steps  0

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

class NodesManager:
...
	def initialize(self):
	...
	startup_nodes_reachable = False
	
	for startup_node in self.startup_nodes.values():
		startup_nodes_reachable = True
	
	...
	if not startup_nodes_reachable:
		raise RedisClusterException(
			f"Redis Cluster cannot be connected. Please provide at least "
			f"one reachable node: {str(exception)}"
		) from exception

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

class RedisCluster(AbstractRedisCluster, RedisClusterCommands):
...
def __init__(
	self,
	host: Optional[str] = None,
	port: int = 6379,
	startup_nodes: Optional[List["ClusterNode"]] = None,
	cluster_error_retry_attempts: int = 3,
	retry: Optional["Retry"] = None,
	require_full_coverage: bool = False,
	reinitialize_steps: int = 5,
	read_from_replicas: bool = False,
	dynamic_startup_nodes: bool = True,
	url: Optional[str] = None,
	**kwargs,
	):
	...
	if url is not None:
		url_options = parse_url(url)
		...
		kwargs.update(url_options)
		host = kwargs.get("host")
		port = kwargs.get("port", port)
		startup_nodes.append(ClusterNode(host, port))
	elif host is not None and port is not None:
		startup_nodes.append(ClusterNode(host, port))
	...
	self.nodes_manager = NodesManager(
		startup_nodes=startup_nodes, # 여기! 
		from_url=from_url,
		require_full_coverage=require_full_coverage,
		dynamic_startup_nodes=dynamic_startup_nodes,
		**kwargs,
	)