Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/redis/sentinel.py: 25%
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 random
2import weakref
3from typing import Optional, Union
5from redis.client import Redis
6from redis.commands import SentinelCommands
7from redis.connection import Connection, ConnectionPool, SSLConnection
8from redis.exceptions import (
9 ConnectionError,
10 ReadOnlyError,
11 ResponseError,
12 TimeoutError,
13)
14from redis.utils import SENTINEL
17class MasterNotFoundError(ConnectionError):
18 pass
21class SlaveNotFoundError(ConnectionError):
22 pass
25ReplicaNotFoundError = SlaveNotFoundError
28class SentinelManagedConnection(Connection):
29 def __init__(self, **kwargs):
30 self.connection_pool = kwargs.pop("connection_pool")
31 super().__init__(**kwargs)
33 def __repr__(self):
34 pool = self.connection_pool
35 s = (
36 f"<{type(self).__module__}.{type(self).__name__}"
37 f"(service={pool.service_name}%s)>"
38 )
39 if self.host:
40 host_info = f",host={self.host},port={self.port}"
41 s = s % host_info
42 return s
44 def connect_to(self, address):
45 self.host, self.port = address
47 self.connect_check_health(
48 check_health=self.connection_pool.check_connection,
49 retry_socket_connect=False,
50 )
52 def _connect_retry(self):
53 if self._sock:
54 return # already connected
55 if self.connection_pool.is_master:
56 self.connect_to(self.connection_pool.get_master_address())
57 else:
58 for slave in self.connection_pool.rotate_slaves():
59 try:
60 return self.connect_to(slave)
61 except ConnectionError:
62 continue
63 raise SlaveNotFoundError # Never be here
65 def connect(self):
66 return self.retry.call_with_retry(self._connect_retry, lambda error: None)
68 def read_response(
69 self,
70 disable_decoding=False,
71 *,
72 timeout: Union[float, object] = SENTINEL,
73 disconnect_on_error: Optional[bool] = False,
74 push_request: Optional[bool] = False,
75 ):
76 try:
77 return super().read_response(
78 disable_decoding=disable_decoding,
79 timeout=timeout,
80 disconnect_on_error=disconnect_on_error,
81 push_request=push_request,
82 )
83 except ReadOnlyError:
84 if self.connection_pool.is_master:
85 # When talking to a master, a ReadOnlyError when likely
86 # indicates that the previous master that we're still connected
87 # to has been demoted to a slave and there's a new master.
88 # calling disconnect will force the connection to re-query
89 # sentinel during the next connect() attempt.
90 self.disconnect()
91 raise ConnectionError("The previous master is now a slave")
92 raise
95class SentinelManagedSSLConnection(SentinelManagedConnection, SSLConnection):
96 pass
99class SentinelConnectionPoolProxy:
100 def __init__(
101 self,
102 connection_pool,
103 is_master,
104 check_connection,
105 service_name,
106 sentinel_manager,
107 ):
108 self.connection_pool_ref = weakref.ref(connection_pool)
109 self.is_master = is_master
110 self.check_connection = check_connection
111 self.service_name = service_name
112 self.sentinel_manager = sentinel_manager
113 self.reset()
115 def reset(self):
116 self.master_address = None
117 self.slave_rr_counter = None
119 def get_master_address(self):
120 master_address = self.sentinel_manager.discover_master(self.service_name)
121 if self.is_master and self.master_address != master_address:
122 self.master_address = master_address
123 # disconnect any idle connections so that they reconnect
124 # to the new master the next time that they are used.
125 connection_pool = self.connection_pool_ref()
126 if connection_pool is not None:
127 connection_pool.disconnect(inuse_connections=False)
128 return master_address
130 def rotate_slaves(self):
131 slaves = self.sentinel_manager.discover_slaves(self.service_name)
132 if slaves:
133 if self.slave_rr_counter is None:
134 self.slave_rr_counter = random.randint(0, len(slaves) - 1)
135 for _ in range(len(slaves)):
136 self.slave_rr_counter = (self.slave_rr_counter + 1) % len(slaves)
137 slave = slaves[self.slave_rr_counter]
138 yield slave
139 # Fallback to the master connection
140 try:
141 yield self.get_master_address()
142 except MasterNotFoundError:
143 pass
144 raise SlaveNotFoundError(f"No slave found for {self.service_name!r}")
146 def rotate_replicas(self):
147 """Round-robin replica balancer.
149 This is an alias for :py:meth:`rotate_slaves`,
150 using the preferred Redis 5.0+ terminology.
151 """
152 return self.rotate_slaves()
155class SentinelConnectionPool(ConnectionPool):
156 """
157 Sentinel backed connection pool.
159 If ``check_connection`` flag is set to True, SentinelManagedConnection
160 sends a PING command right after establishing the connection.
161 """
163 def __init__(self, service_name, sentinel_manager, **kwargs):
164 kwargs["connection_class"] = kwargs.get(
165 "connection_class",
166 (
167 SentinelManagedSSLConnection
168 if kwargs.pop("ssl", False)
169 else SentinelManagedConnection
170 ),
171 )
172 self.is_master = kwargs.pop("is_master", True)
173 self.check_connection = kwargs.pop("check_connection", False)
174 self.proxy = SentinelConnectionPoolProxy(
175 connection_pool=self,
176 is_master=self.is_master,
177 check_connection=self.check_connection,
178 service_name=service_name,
179 sentinel_manager=sentinel_manager,
180 )
181 super().__init__(**kwargs)
182 self.connection_kwargs["connection_pool"] = self.proxy
183 self.service_name = service_name
184 self.sentinel_manager = sentinel_manager
186 def __repr__(self):
187 role = "master" if self.is_master else "slave"
188 return (
189 f"<{type(self).__module__}.{type(self).__name__}"
190 f"(service={self.service_name}({role}))>"
191 )
193 def reset(self):
194 super().reset()
195 self.proxy.reset()
197 @property
198 def master_address(self):
199 return self.proxy.master_address
201 def owns_connection(self, connection):
202 check = not self.is_master or (
203 self.is_master and self.master_address == (connection.host, connection.port)
204 )
205 parent = super()
206 return check and parent.owns_connection(connection)
208 def get_master_address(self):
209 return self.proxy.get_master_address()
211 def rotate_slaves(self):
212 "Round-robin slave balancer"
213 return self.proxy.rotate_slaves()
215 def rotate_replicas(self):
216 """Round-robin replica balancer.
218 This is an alias for :py:meth:`rotate_slaves`,
219 using the preferred Redis 5.0+ terminology.
220 """
221 return self.rotate_slaves()
224class Sentinel(SentinelCommands):
225 """
226 Redis Sentinel cluster client
228 >>> from redis.sentinel import Sentinel
229 >>> sentinel = Sentinel([('localhost', 26379)], socket_timeout=0.1)
230 >>> master = sentinel.master_for('mymaster', socket_timeout=0.1)
231 >>> master.set('foo', 'bar')
232 >>> slave = sentinel.slave_for('mymaster', socket_timeout=0.1)
233 >>> slave.get('foo')
234 b'bar'
236 ``sentinels`` is a list of sentinel nodes. Each node is represented by
237 a pair (hostname, port).
239 ``min_other_sentinels`` defined a minimum number of peers for a sentinel.
240 When querying a sentinel, if it doesn't meet this threshold, responses
241 from that sentinel won't be considered valid.
243 ``sentinel_kwargs`` is a dictionary of connection arguments used when
244 connecting to sentinel instances. Any argument that can be passed to
245 a normal Redis connection can be specified here. If ``sentinel_kwargs`` is
246 not specified, any socket_timeout and socket_keepalive options specified
247 in ``connection_kwargs`` will be used.
249 ``connection_kwargs`` are keyword arguments that will be used when
250 establishing a connection to a Redis server.
251 """
253 def __init__(
254 self,
255 sentinels,
256 min_other_sentinels=0,
257 sentinel_kwargs=None,
258 force_master_ip=None,
259 **connection_kwargs,
260 ):
261 # if sentinel_kwargs isn't defined, use the socket_* options from
262 # connection_kwargs
263 if sentinel_kwargs is None:
264 sentinel_kwargs = {
265 k: v for k, v in connection_kwargs.items() if k.startswith("socket_")
266 }
267 self.sentinel_kwargs = sentinel_kwargs
269 self.sentinels = [
270 Redis(hostname, port, **self.sentinel_kwargs)
271 for hostname, port in sentinels
272 ]
273 self.min_other_sentinels = min_other_sentinels
274 self.connection_kwargs = connection_kwargs
275 self._force_master_ip = force_master_ip
277 def execute_command(self, *args, **kwargs):
278 """
279 Execute Sentinel command in sentinel nodes.
280 once - If set to True, then execute the resulting command on a single
281 node at random, rather than across the entire sentinel cluster.
282 """
283 once = bool(kwargs.pop("once", False))
285 # Check if command is supposed to return the original
286 # responses instead of boolean value.
287 return_responses = bool(kwargs.pop("return_responses", False))
289 if once:
290 response = random.choice(self.sentinels).execute_command(*args, **kwargs)
291 if return_responses:
292 return [response]
293 else:
294 return True if response else False
296 responses = []
297 for sentinel in self.sentinels:
298 responses.append(sentinel.execute_command(*args, **kwargs))
300 if return_responses:
301 return responses
303 return all(responses)
305 def __repr__(self):
306 sentinel_addresses = []
307 for sentinel in self.sentinels:
308 sentinel_addresses.append(
309 "{host}:{port}".format_map(sentinel.connection_pool.connection_kwargs)
310 )
311 return (
312 f"<{type(self).__module__}.{type(self).__name__}"
313 f"(sentinels=[{','.join(sentinel_addresses)}])>"
314 )
316 def close(self) -> None:
317 """
318 Close all sentinel clients created by this Sentinel and their
319 connection pools.
321 Each client is closed independently: if one raises, the remaining
322 clients are still closed and the first error is re-raised afterwards.
324 Clients returned by ``master_for``/``slave_for`` are owned by the
325 caller and are not closed here.
326 """
327 exc = None
328 for sentinel in self.sentinels:
329 try:
330 sentinel.close()
331 except Exception as e:
332 if exc is None:
333 exc = e
334 if exc:
335 raise exc
337 def __enter__(self):
338 return self
340 def __exit__(self, exc_type, exc_value, traceback):
341 self.close()
343 def check_master_state(self, state, service_name):
344 if not state["is_master"] or state["is_sdown"] or state["is_odown"]:
345 return False
346 # Check if our sentinel doesn't see other nodes
347 if state["num-other-sentinels"] < self.min_other_sentinels:
348 return False
349 return True
351 def discover_master(self, service_name):
352 """
353 Asks sentinel servers for the Redis master's address corresponding
354 to the service labeled ``service_name``.
356 Returns a pair (address, port) or raises MasterNotFoundError if no
357 master is found.
358 """
359 collected_errors = list()
360 for sentinel_no, sentinel in enumerate(self.sentinels):
361 try:
362 masters = sentinel.sentinel_masters()
363 except (ConnectionError, TimeoutError) as e:
364 collected_errors.append(f"{sentinel} - {e!r}")
365 continue
366 state = masters.get(service_name)
367 if state and self.check_master_state(state, service_name):
368 # Put this sentinel at the top of the list
369 self.sentinels[0], self.sentinels[sentinel_no] = (
370 sentinel,
371 self.sentinels[0],
372 )
374 ip = (
375 self._force_master_ip
376 if self._force_master_ip is not None
377 else state["ip"]
378 )
379 return ip, state["port"]
381 error_info = ""
382 if len(collected_errors) > 0:
383 error_info = f" : {', '.join(collected_errors)}"
384 raise MasterNotFoundError(f"No master found for {service_name!r}{error_info}")
386 def filter_slaves(self, slaves):
387 "Remove slaves that are in an ODOWN or SDOWN state"
388 slaves_alive = []
389 for slave in slaves:
390 if slave["is_odown"] or slave["is_sdown"]:
391 continue
392 slaves_alive.append((slave["ip"], slave["port"]))
393 return slaves_alive
395 def filter_replicas(self, replicas):
396 """Remove replicas that are in an ODOWN or SDOWN state.
398 This is an alias for :py:meth:`filter_slaves`,
399 using the preferred Redis 5.0+ terminology.
400 """
401 return self.filter_slaves(replicas)
403 def discover_slaves(self, service_name):
404 "Returns a list of alive slaves for service ``service_name``"
405 for sentinel in self.sentinels:
406 try:
407 slaves = sentinel.sentinel_slaves(service_name)
408 except (ConnectionError, ResponseError, TimeoutError):
409 continue
410 slaves = self.filter_slaves(slaves)
411 if slaves:
412 return slaves
413 return []
415 def discover_replicas(self, service_name):
416 """Returns a list of alive replicas for service ``service_name``.
418 This is an alias for :py:meth:`discover_slaves`,
419 using the preferred Redis 5.0+ terminology.
420 """
421 return self.discover_slaves(service_name)
423 def master_for(
424 self,
425 service_name,
426 redis_class=Redis,
427 connection_pool_class=SentinelConnectionPool,
428 **kwargs,
429 ):
430 """
431 Returns a redis client instance for the ``service_name`` master.
432 Sentinel client will detect failover and reconnect Redis clients
433 automatically.
435 A :py:class:`~redis.sentinel.SentinelConnectionPool` class is
436 used to retrieve the master's address before establishing a new
437 connection.
439 NOTE: If the master's address has changed, any cached connections to
440 the old master are closed.
442 By default clients will be a :py:class:`~redis.Redis` instance.
443 Specify a different class to the ``redis_class`` argument if you
444 desire something different.
446 The ``connection_pool_class`` specifies the connection pool to
447 use. The :py:class:`~redis.sentinel.SentinelConnectionPool`
448 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"] = True
455 connection_kwargs = dict(self.connection_kwargs)
456 connection_kwargs.update(kwargs)
457 return redis_class.from_pool(
458 connection_pool_class(service_name, self, **connection_kwargs)
459 )
461 def slave_for(
462 self,
463 service_name,
464 redis_class=Redis,
465 connection_pool_class=SentinelConnectionPool,
466 **kwargs,
467 ):
468 """
469 Returns redis client instance for the ``service_name`` slave(s).
471 A SentinelConnectionPool class is used to retrieve the slave's
472 address before establishing a new connection.
474 By default clients will be a :py:class:`~redis.Redis` instance.
475 Specify a different class to the ``redis_class`` argument if you
476 desire something different.
478 The ``connection_pool_class`` specifies the connection pool to use.
479 The SentinelConnectionPool will be used by default.
481 All other keyword arguments are merged with any connection_kwargs
482 passed to this class and passed to the connection pool as keyword
483 arguments to be used to initialize Redis connections.
484 """
485 kwargs["is_master"] = False
486 connection_kwargs = dict(self.connection_kwargs)
487 connection_kwargs.update(kwargs)
488 return redis_class.from_pool(
489 connection_pool_class(service_name, self, **connection_kwargs)
490 )
492 def replica_for(
493 self,
494 service_name,
495 redis_class=Redis,
496 connection_pool_class=SentinelConnectionPool,
497 **kwargs,
498 ):
499 """
500 Returns redis client instance for the ``service_name`` replica(s).
502 This is an alias for :py:meth:`slave_for`,
503 using the preferred Redis 5.0+ terminology.
504 """
505 return self.slave_for(
506 service_name,
507 redis_class=redis_class,
508 connection_pool_class=connection_pool_class,
509 **kwargs,
510 )