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 asyncio
2import random
3import weakref
4from typing import AsyncIterator, Iterable, Mapping, Optional, Sequence, Tuple, Type
6from redis.asyncio.client import Redis
7from redis.asyncio.connection import (
8 Connection,
9 ConnectionPool,
10 EncodableT,
11 SSLConnection,
12)
13from redis.commands import AsyncSentinelCommands
14from redis.exceptions import (
15 ConnectionError,
16 ReadOnlyError,
17 ResponseError,
18 TimeoutError,
19)
22class MasterNotFoundError(ConnectionError):
23 pass
26class SlaveNotFoundError(ConnectionError):
27 pass
30ReplicaNotFoundError = SlaveNotFoundError
33class SentinelManagedConnection(Connection):
34 def __init__(self, **kwargs):
35 self.connection_pool = kwargs.pop("connection_pool")
36 super().__init__(**kwargs)
38 def __repr__(self):
39 s = f"<{self.__class__.__module__}.{self.__class__.__name__}"
40 if self.host:
41 host_info = f",host={self.host},port={self.port}"
42 s += host_info
43 return s + ")>"
45 async def connect_to(self, address):
46 self.host, self.port = address
47 await self.connect_check_health(
48 check_health=self.connection_pool.check_connection,
49 retry_socket_connect=False,
50 )
52 async def _connect_retry(self):
53 if self._reader:
54 return # already connected
55 if self.connection_pool.is_master:
56 await self.connect_to(await self.connection_pool.get_master_address())
57 else:
58 async for slave in self.connection_pool.rotate_slaves():
59 try:
60 return await self.connect_to(slave)
61 except ConnectionError:
62 continue
63 raise SlaveNotFoundError # Never be here
65 async def connect(self):
66 return await self.retry.call_with_retry(
67 self._connect_retry,
68 lambda error: asyncio.sleep(0),
69 )
71 async def read_response(
72 self,
73 disable_decoding: bool = False,
74 timeout: Optional[float] = None,
75 *,
76 disconnect_on_error: Optional[float] = True,
77 push_request: Optional[bool] = False,
78 ):
79 try:
80 return await super().read_response(
81 disable_decoding=disable_decoding,
82 timeout=timeout,
83 disconnect_on_error=disconnect_on_error,
84 push_request=push_request,
85 )
86 except ReadOnlyError:
87 if self.connection_pool.is_master:
88 # When talking to a master, a ReadOnlyError when likely
89 # indicates that the previous master that we're still connected
90 # to has been demoted to a slave and there's a new master.
91 # calling disconnect will force the connection to re-query
92 # sentinel during the next connect() attempt.
93 await self.disconnect()
94 raise ConnectionError("The previous master is now a slave")
95 raise
98class SentinelManagedSSLConnection(SentinelManagedConnection, SSLConnection):
99 pass
102class SentinelConnectionPool(ConnectionPool):
103 """
104 Sentinel backed connection pool.
106 If ``check_connection`` flag is set to True, SentinelManagedConnection
107 sends a PING command right after establishing the connection.
108 """
110 def __init__(self, service_name, sentinel_manager, **kwargs):
111 kwargs["connection_class"] = kwargs.get(
112 "connection_class",
113 (
114 SentinelManagedSSLConnection
115 if kwargs.pop("ssl", False)
116 else SentinelManagedConnection
117 ),
118 )
119 self.is_master = kwargs.pop("is_master", True)
120 self.check_connection = kwargs.pop("check_connection", False)
121 super().__init__(**kwargs)
122 self.connection_kwargs["connection_pool"] = weakref.proxy(self)
123 self.service_name = service_name
124 self.sentinel_manager = sentinel_manager
125 self.master_address = None
126 self.slave_rr_counter = None
128 def __repr__(self):
129 return (
130 f"<{self.__class__.__module__}.{self.__class__.__name__}"
131 f"(service={self.service_name}({self.is_master and 'master' or 'slave'}))>"
132 )
134 def reset(self):
135 super().reset()
136 self.master_address = None
137 self.slave_rr_counter = None
139 def owns_connection(self, connection: Connection):
140 check = not self.is_master or (
141 self.is_master and self.master_address == (connection.host, connection.port)
142 )
143 return check and super().owns_connection(connection)
145 async def get_master_address(self):
146 master_address = await self.sentinel_manager.discover_master(self.service_name)
147 if self.is_master:
148 if self.master_address != master_address:
149 self.master_address = master_address
150 # disconnect any idle connections so that they reconnect
151 # to the new master the next time that they are used.
152 await self.disconnect(inuse_connections=False)
153 return master_address
155 async def rotate_slaves(self) -> AsyncIterator:
156 """Round-robin slave balancer"""
157 slaves = await self.sentinel_manager.discover_slaves(self.service_name)
158 if slaves:
159 if self.slave_rr_counter is None:
160 self.slave_rr_counter = random.randint(0, len(slaves) - 1)
161 for _ in range(len(slaves)):
162 self.slave_rr_counter = (self.slave_rr_counter + 1) % len(slaves)
163 slave = slaves[self.slave_rr_counter]
164 yield slave
165 # Fallback to the master connection
166 try:
167 yield await self.get_master_address()
168 except MasterNotFoundError:
169 pass
170 raise SlaveNotFoundError(f"No slave found for {self.service_name!r}")
172 def rotate_replicas(self) -> AsyncIterator:
173 """Round-robin replica balancer.
175 This is an alias for :py:meth:`rotate_slaves`,
176 using the preferred Redis 5.0+ terminology.
177 """
178 return self.rotate_slaves()
181class Sentinel(AsyncSentinelCommands):
182 """
183 Redis Sentinel cluster client
185 >>> from redis.sentinel import Sentinel
186 >>> sentinel = Sentinel([('localhost', 26379)], socket_timeout=0.1)
187 >>> master = sentinel.master_for('mymaster', socket_timeout=0.1)
188 >>> await master.set('foo', 'bar')
189 >>> slave = sentinel.slave_for('mymaster', socket_timeout=0.1)
190 >>> await slave.get('foo')
191 b'bar'
193 ``sentinels`` is a list of sentinel nodes. Each node is represented by
194 a pair (hostname, port).
196 ``min_other_sentinels`` defined a minimum number of peers for a sentinel.
197 When querying a sentinel, if it doesn't meet this threshold, responses
198 from that sentinel won't be considered valid.
200 ``sentinel_kwargs`` is a dictionary of connection arguments used when
201 connecting to sentinel instances. Any argument that can be passed to
202 a normal Redis connection can be specified here. If ``sentinel_kwargs`` is
203 not specified, any socket_timeout and socket_keepalive options specified
204 in ``connection_kwargs`` will be used.
206 ``connection_kwargs`` are keyword arguments that will be used when
207 establishing a connection to a Redis server.
208 """
210 def __init__(
211 self,
212 sentinels,
213 min_other_sentinels=0,
214 sentinel_kwargs=None,
215 force_master_ip=None,
216 **connection_kwargs,
217 ):
218 # if sentinel_kwargs isn't defined, use the socket_* options from
219 # connection_kwargs
220 if sentinel_kwargs is None:
221 sentinel_kwargs = {
222 k: v for k, v in connection_kwargs.items() if k.startswith("socket_")
223 }
224 self.sentinel_kwargs = sentinel_kwargs
226 self.sentinels = [
227 Redis(host=hostname, port=port, **self.sentinel_kwargs)
228 for hostname, port in sentinels
229 ]
230 self.min_other_sentinels = min_other_sentinels
231 self.connection_kwargs = connection_kwargs
232 self._force_master_ip = force_master_ip
234 async def execute_command(self, *args, **kwargs):
235 """
236 Execute Sentinel command in sentinel nodes.
237 once - If set to True, then execute the resulting command on a single
238 node at random, rather than across the entire sentinel cluster.
239 """
240 once = bool(kwargs.pop("once", False))
242 # Check if command is supposed to return the original
243 # responses instead of boolean value.
244 return_responses = bool(kwargs.pop("return_responses", False))
246 if once:
247 response = await random.choice(self.sentinels).execute_command(
248 *args, **kwargs
249 )
250 if return_responses:
251 return [response]
252 else:
253 return True if response else False
255 tasks = [
256 asyncio.Task(sentinel.execute_command(*args, **kwargs))
257 for sentinel in self.sentinels
258 ]
259 responses = await asyncio.gather(*tasks)
261 if return_responses:
262 return responses
264 return all(responses)
266 def __repr__(self):
267 sentinel_addresses = []
268 for sentinel in self.sentinels:
269 sentinel_addresses.append(
270 f"{sentinel.connection_pool.connection_kwargs['host']}:"
271 f"{sentinel.connection_pool.connection_kwargs['port']}"
272 )
273 return (
274 f"<{self.__class__}.{self.__class__.__name__}"
275 f"(sentinels=[{','.join(sentinel_addresses)}])>"
276 )
278 async def aclose(self) -> None:
279 """
280 Close all sentinel clients created by this Sentinel and their
281 connection pools.
283 Each client is closed independently: if one raises, the remaining
284 clients are still closed and the first error is re-raised afterwards.
286 Clients returned by ``master_for``/``slave_for`` are owned by the
287 caller and are not closed here.
288 """
289 resp = await asyncio.gather(
290 *(sentinel.aclose() for sentinel in self.sentinels),
291 return_exceptions=True,
292 )
293 exc = next((r for r in resp if isinstance(r, BaseException)), None)
294 if exc:
295 raise exc
297 async def __aenter__(self) -> "Sentinel":
298 return self
300 async def __aexit__(self, exc_type, exc_value, traceback) -> None:
301 await self.aclose()
303 def check_master_state(self, state: dict, service_name: str) -> bool:
304 if not state["is_master"] or state["is_sdown"] or state["is_odown"]:
305 return False
306 # Check if our sentinel doesn't see other nodes
307 if state["num-other-sentinels"] < self.min_other_sentinels:
308 return False
309 return True
311 async def discover_master(self, service_name: str):
312 """
313 Asks sentinel servers for the Redis master's address corresponding
314 to the service labeled ``service_name``.
316 Returns a pair (address, port) or raises MasterNotFoundError if no
317 master is found.
318 """
319 collected_errors = list()
320 for sentinel_no, sentinel in enumerate(self.sentinels):
321 try:
322 masters = await sentinel.sentinel_masters()
323 except (ConnectionError, TimeoutError) as e:
324 collected_errors.append(f"{sentinel} - {e!r}")
325 continue
326 state = masters.get(service_name)
327 if state and self.check_master_state(state, service_name):
328 # Put this sentinel at the top of the list
329 self.sentinels[0], self.sentinels[sentinel_no] = (
330 sentinel,
331 self.sentinels[0],
332 )
334 ip = (
335 self._force_master_ip
336 if self._force_master_ip is not None
337 else state["ip"]
338 )
339 return ip, state["port"]
341 error_info = ""
342 if len(collected_errors) > 0:
343 error_info = f" : {', '.join(collected_errors)}"
344 raise MasterNotFoundError(f"No master found for {service_name!r}{error_info}")
346 def filter_slaves(
347 self, slaves: Iterable[Mapping]
348 ) -> Sequence[Tuple[EncodableT, EncodableT]]:
349 """Remove slaves that are in an ODOWN or SDOWN state"""
350 slaves_alive = []
351 for slave in slaves:
352 if slave["is_odown"] or slave["is_sdown"]:
353 continue
354 slaves_alive.append((slave["ip"], slave["port"]))
355 return slaves_alive
357 def filter_replicas(
358 self, replicas: Iterable[Mapping]
359 ) -> Sequence[Tuple[EncodableT, EncodableT]]:
360 """Remove replicas that are in an ODOWN or SDOWN state.
362 This is an alias for :py:meth:`filter_slaves`,
363 using the preferred Redis 5.0+ terminology.
364 """
365 return self.filter_slaves(replicas)
367 async def discover_slaves(
368 self, service_name: str
369 ) -> Sequence[Tuple[EncodableT, EncodableT]]:
370 """Returns a list of alive slaves for service ``service_name``"""
371 for sentinel in self.sentinels:
372 try:
373 slaves = await sentinel.sentinel_slaves(service_name)
374 except (ConnectionError, ResponseError, TimeoutError):
375 continue
376 slaves = self.filter_slaves(slaves)
377 if slaves:
378 return slaves
379 return []
381 async def discover_replicas(
382 self, service_name: str
383 ) -> Sequence[Tuple[EncodableT, EncodableT]]:
384 """Returns a list of alive replicas for service ``service_name``.
386 This is an alias for :py:meth:`discover_slaves`,
387 using the preferred Redis 5.0+ terminology.
388 """
389 return await self.discover_slaves(service_name)
391 def master_for(
392 self,
393 service_name: str,
394 redis_class: Type[Redis] = Redis,
395 connection_pool_class: Type[SentinelConnectionPool] = SentinelConnectionPool,
396 **kwargs,
397 ):
398 """
399 Returns a redis client instance for the ``service_name`` master.
400 Sentinel client will detect failover and reconnect Redis clients
401 automatically.
403 A :py:class:`~redis.sentinel.SentinelConnectionPool` class is
404 used to retrieve the master's address before establishing a new
405 connection.
407 NOTE: If the master's address has changed, any cached connections to
408 the old master are closed.
410 By default clients will be a :py:class:`~redis.Redis` instance.
411 Specify a different class to the ``redis_class`` argument if you
412 desire something different.
414 The ``connection_pool_class`` specifies the connection pool to
415 use. The :py:class:`~redis.sentinel.SentinelConnectionPool`
416 will be used by default.
418 All other keyword arguments are merged with any connection_kwargs
419 passed to this class and passed to the connection pool as keyword
420 arguments to be used to initialize Redis connections.
421 """
422 kwargs["is_master"] = True
423 connection_kwargs = dict(self.connection_kwargs)
424 connection_kwargs.update(kwargs)
426 connection_pool = connection_pool_class(service_name, self, **connection_kwargs)
427 # The Redis object "owns" the pool
428 return redis_class.from_pool(connection_pool)
430 def slave_for(
431 self,
432 service_name: str,
433 redis_class: Type[Redis] = Redis,
434 connection_pool_class: Type[SentinelConnectionPool] = SentinelConnectionPool,
435 **kwargs,
436 ):
437 """
438 Returns redis client instance for the ``service_name`` slave(s).
440 A SentinelConnectionPool class is used to retrieve the slave's
441 address before establishing a new connection.
443 By default clients will be a :py:class:`~redis.Redis` instance.
444 Specify a different class to the ``redis_class`` argument if you
445 desire something different.
447 The ``connection_pool_class`` specifies the connection pool to use.
448 The SentinelConnectionPool will be used by default.
450 All other keyword arguments are merged with any connection_kwargs
451 passed to this class and passed to the connection pool as keyword
452 arguments to be used to initialize Redis connections.
453 """
454 kwargs["is_master"] = False
455 connection_kwargs = dict(self.connection_kwargs)
456 connection_kwargs.update(kwargs)
458 connection_pool = connection_pool_class(service_name, self, **connection_kwargs)
459 # The Redis object "owns" the pool
460 return redis_class.from_pool(connection_pool)
462 def replica_for(
463 self,
464 service_name: str,
465 redis_class: Type[Redis] = Redis,
466 connection_pool_class: Type[SentinelConnectionPool] = SentinelConnectionPool,
467 **kwargs,
468 ):
469 """
470 Returns redis client instance for the ``service_name`` replica(s).
472 This is an alias for :py:meth:`slave_for`,
473 using the preferred Redis 5.0+ terminology.
474 """
475 return self.slave_for(
476 service_name,
477 redis_class=redis_class,
478 connection_pool_class=connection_pool_class,
479 **kwargs,
480 )