import asyncio
|
import logging
|
from abc import ABC, abstractmethod
|
from enum import Enum
|
from typing import List, Optional, Tuple, Union
|
|
from redis.asyncio import Redis
|
from redis.asyncio.http.http_client import DEFAULT_TIMEOUT, AsyncHTTPClientWrapper
|
from redis.backoff import NoBackoff
|
from redis.http.http_client import HttpClient
|
from redis.multidb.exception import UnhealthyDatabaseException
|
from redis.retry import Retry
|
|
DEFAULT_HEALTH_CHECK_PROBES = 3
|
DEFAULT_HEALTH_CHECK_INTERVAL = 5
|
DEFAULT_HEALTH_CHECK_DELAY = 0.5
|
DEFAULT_LAG_AWARE_TOLERANCE = 5000
|
|
logger = logging.getLogger(__name__)
|
|
|
class HealthCheck(ABC):
|
@abstractmethod
|
async def check_health(self, database) -> bool:
|
"""Function to determine the health status."""
|
pass
|
|
|
class HealthCheckPolicy(ABC):
|
"""
|
Health checks execution policy.
|
"""
|
|
@property
|
@abstractmethod
|
def health_check_probes(self) -> int:
|
"""Number of probes to execute health checks."""
|
pass
|
|
@property
|
@abstractmethod
|
def health_check_delay(self) -> float:
|
"""Delay between health check probes."""
|
pass
|
|
@abstractmethod
|
async def execute(self, health_checks: List[HealthCheck], database) -> bool:
|
"""Execute health checks and return database health status."""
|
pass
|
|
|
class AbstractHealthCheckPolicy(HealthCheckPolicy):
|
def __init__(self, health_check_probes: int, health_check_delay: float):
|
if health_check_probes < 1:
|
raise ValueError("health_check_probes must be greater than 0")
|
self._health_check_probes = health_check_probes
|
self._health_check_delay = health_check_delay
|
|
@property
|
def health_check_probes(self) -> int:
|
return self._health_check_probes
|
|
@property
|
def health_check_delay(self) -> float:
|
return self._health_check_delay
|
|
@abstractmethod
|
async def execute(self, health_checks: List[HealthCheck], database) -> bool:
|
pass
|
|
|
class HealthyAllPolicy(AbstractHealthCheckPolicy):
|
"""
|
Policy that returns True if all health check probes are successful.
|
"""
|
|
def __init__(self, health_check_probes: int, health_check_delay: float):
|
super().__init__(health_check_probes, health_check_delay)
|
|
async def execute(self, health_checks: List[HealthCheck], database) -> bool:
|
for health_check in health_checks:
|
for attempt in range(self.health_check_probes):
|
try:
|
if not await health_check.check_health(database):
|
return False
|
except Exception as e:
|
raise UnhealthyDatabaseException("Unhealthy database", database, e)
|
|
if attempt < self.health_check_probes - 1:
|
await asyncio.sleep(self._health_check_delay)
|
return True
|
|
|
class HealthyMajorityPolicy(AbstractHealthCheckPolicy):
|
"""
|
Policy that returns True if a majority of health check probes are successful.
|
"""
|
|
def __init__(self, health_check_probes: int, health_check_delay: float):
|
super().__init__(health_check_probes, health_check_delay)
|
|
async def execute(self, health_checks: List[HealthCheck], database) -> bool:
|
for health_check in health_checks:
|
if self.health_check_probes % 2 == 0:
|
allowed_unsuccessful_probes = self.health_check_probes / 2
|
else:
|
allowed_unsuccessful_probes = (self.health_check_probes + 1) / 2
|
|
for attempt in range(self.health_check_probes):
|
try:
|
if not await health_check.check_health(database):
|
allowed_unsuccessful_probes -= 1
|
if allowed_unsuccessful_probes <= 0:
|
return False
|
except Exception as e:
|
allowed_unsuccessful_probes -= 1
|
if allowed_unsuccessful_probes <= 0:
|
raise UnhealthyDatabaseException(
|
"Unhealthy database", database, e
|
)
|
|
if attempt < self.health_check_probes - 1:
|
await asyncio.sleep(self._health_check_delay)
|
return True
|
|
|
class HealthyAnyPolicy(AbstractHealthCheckPolicy):
|
"""
|
Policy that returns True if at least one health check probe is successful.
|
"""
|
|
def __init__(self, health_check_probes: int, health_check_delay: float):
|
super().__init__(health_check_probes, health_check_delay)
|
|
async def execute(self, health_checks: List[HealthCheck], database) -> bool:
|
is_healthy = False
|
|
for health_check in health_checks:
|
exception = None
|
|
for attempt in range(self.health_check_probes):
|
try:
|
if await health_check.check_health(database):
|
is_healthy = True
|
break
|
else:
|
is_healthy = False
|
except Exception as e:
|
exception = UnhealthyDatabaseException(
|
"Unhealthy database", database, e
|
)
|
|
if attempt < self.health_check_probes - 1:
|
await asyncio.sleep(self._health_check_delay)
|
|
if not is_healthy and not exception:
|
return is_healthy
|
elif not is_healthy and exception:
|
raise exception
|
|
return is_healthy
|
|
|
class HealthCheckPolicies(Enum):
|
HEALTHY_ALL = HealthyAllPolicy
|
HEALTHY_MAJORITY = HealthyMajorityPolicy
|
HEALTHY_ANY = HealthyAnyPolicy
|
|
|
DEFAULT_HEALTH_CHECK_POLICY: HealthCheckPolicies = HealthCheckPolicies.HEALTHY_ALL
|
|
|
class PingHealthCheck(HealthCheck):
|
"""
|
Health check based on PING command.
|
"""
|
|
async def check_health(self, database) -> bool:
|
if isinstance(database.client, Redis):
|
return await database.client.execute_command("PING")
|
else:
|
# For a cluster checks if all nodes are healthy.
|
all_nodes = database.client.get_nodes()
|
for node in all_nodes:
|
if not await node.redis_connection.execute_command("PING"):
|
return False
|
|
return True
|
|
|
class LagAwareHealthCheck(HealthCheck):
|
"""
|
Health check available for Redis Enterprise deployments.
|
Verify via REST API that the database is healthy based on different lags.
|
"""
|
|
def __init__(
|
self,
|
rest_api_port: int = 9443,
|
lag_aware_tolerance: int = DEFAULT_LAG_AWARE_TOLERANCE,
|
timeout: float = DEFAULT_TIMEOUT,
|
auth_basic: Optional[Tuple[str, str]] = None,
|
verify_tls: bool = True,
|
# TLS verification (server) options
|
ca_file: Optional[str] = None,
|
ca_path: Optional[str] = None,
|
ca_data: Optional[Union[str, bytes]] = None,
|
# Mutual TLS (client cert) options
|
client_cert_file: Optional[str] = None,
|
client_key_file: Optional[str] = None,
|
client_key_password: Optional[str] = None,
|
):
|
"""
|
Initialize LagAwareHealthCheck with the specified parameters.
|
|
Args:
|
rest_api_port: Port number for Redis Enterprise REST API (default: 9443)
|
lag_aware_tolerance: Tolerance in lag between databases in MS (default: 100)
|
timeout: Request timeout in seconds (default: DEFAULT_TIMEOUT)
|
auth_basic: Tuple of (username, password) for basic authentication
|
verify_tls: Whether to verify TLS certificates (default: True)
|
ca_file: Path to CA certificate file for TLS verification
|
ca_path: Path to CA certificates directory for TLS verification
|
ca_data: CA certificate data as string or bytes
|
client_cert_file: Path to client certificate file for mutual TLS
|
client_key_file: Path to client private key file for mutual TLS
|
client_key_password: Password for encrypted client private key
|
"""
|
self._http_client = AsyncHTTPClientWrapper(
|
HttpClient(
|
timeout=timeout,
|
auth_basic=auth_basic,
|
retry=Retry(NoBackoff(), retries=0),
|
verify_tls=verify_tls,
|
ca_file=ca_file,
|
ca_path=ca_path,
|
ca_data=ca_data,
|
client_cert_file=client_cert_file,
|
client_key_file=client_key_file,
|
client_key_password=client_key_password,
|
)
|
)
|
self._rest_api_port = rest_api_port
|
self._lag_aware_tolerance = lag_aware_tolerance
|
|
async def check_health(self, database) -> bool:
|
if database.health_check_url is None:
|
raise ValueError(
|
"Database health check url is not set. Please check DatabaseConfig for the current database."
|
)
|
|
if isinstance(database.client, Redis):
|
db_host = database.client.get_connection_kwargs()["host"]
|
else:
|
db_host = database.client.startup_nodes[0].host
|
|
base_url = f"{database.health_check_url}:{self._rest_api_port}"
|
self._http_client.client.base_url = base_url
|
|
# Find bdb matching to the current database host
|
matching_bdb = None
|
for bdb in await self._http_client.get("/v1/bdbs"):
|
for endpoint in bdb["endpoints"]:
|
if endpoint["dns_name"] == db_host:
|
matching_bdb = bdb
|
break
|
|
# In case if the host was set as public IP
|
for addr in endpoint["addr"]:
|
if addr == db_host:
|
matching_bdb = bdb
|
break
|
|
if matching_bdb is None:
|
logger.warning("LagAwareHealthCheck failed: Couldn't find a matching bdb")
|
raise ValueError("Could not find a matching bdb")
|
|
url = (
|
f"/v1/bdbs/{matching_bdb['uid']}/availability"
|
f"?extend_check=lag&availability_lag_tolerance_ms={self._lag_aware_tolerance}"
|
)
|
await self._http_client.get(url, expect_json=False)
|
|
# Status checked in an http client, otherwise HttpError will be raised
|
return True
|
