File Content: redis_lib.py # -*- coding: utf-8 -*- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT from __future__ import absolute_import import subprocess from typing import Optional, Tuple, Dict from clwpos import gettext as _ from clcommon.utils import ExternalProgramFailed, run_command from clwpos.cl_wpos_exceptions import WposError class DeadRedisPurgeAttempt(WposError): """ Happens when somebody tries to purge redis instance which is not runnning """ def __init__(self): super().__init__(message=_( "Unable to purge cache because cache database is not running. " "Enable at least one optimization feature to start cache database instance.")) class PurgeFailedError(WposError): """ Happens when redis is not able to purge cached data for some reason. """ def __init__(self, std_out: str, std_err: str): super().__init__( message=_('Unable to purge cache. \n' 'Database backend returned error of command "%(command)s" execution. ' 'Try again a bit later or contact your system administrator for help.'), details='stdout: \n<i>%(stdout)s</i>\n' 'stderr: \n<i>%(stderr)s</i>\n', context={ "command": 'purge', "stdout": std_out or 'empty', "stderr": std_err or 'empty', }) class RedisLibUser(object): def __init__(self, socket_path: str, redis_cli_path: str = "/opt/alt/redis/bin/redis-cli") -> None: self.socket_path = socket_path self.redis_cli_path = redis_cli_path def get_redis_shell_command(self, redis_command): return [ self.redis_cli_path, "-s", self.socket_path, ] + redis_command.split(" ") def run_redis_command(self, command: str) -> Tuple[int, str, str]: redis_command = self.get_redis_shell_command(command) try: returncode, std_out, std_err = run_command( redis_command, return_full_output=True ) except (subprocess.CalledProcessError, ExternalProgramFailed) as error: raise WposError( message=_("Error during %(command)s command execution: \n%(error)s"), context={"command": " ".join(redis_command), "error": str(error)}, ) return returncode, std_out, std_err def purge_redis(self) -> Dict: """ Clean entire redis cache for user. """ if not self.is_redis_alive(): raise DeadRedisPurgeAttempt() purge_command = "flushall async" returncode, std_out, std_err = self.run_redis_command(purge_command) # Output of Redis FLUSHALL command should be: "OK" if returncode != 0 or "ok" not in std_out.lower(): raise PurgeFailedError(std_out, std_err) return { "used_memory": self.get_redis_used_memory() } def is_redis_alive(self) -> bool: """ Check if user's redis is alive. """ returncode, std_out, _ = self.run_redis_command("ping") return returncode == 0 and "pong" in std_out.lower() def is_redis_empty(self) -> bool: """ Check if user's redis is empty. Example (redis is empty): # Keyspace Example (redis is NOT empty): # Keyspace db0:keys=2,expires=0,avg_ttl=0 """ returncode, std_out, _ = self.run_redis_command("info keyspace") return returncode == 0 and len(std_out.strip().split("\n")) <= 1 def get_redis_used_memory(self) -> Optional[str]: """ Return amount of memmory used by user's redis instance in human readable format (kb or mb). If redis status is offline, 'used_memory_value' value is null. 'info memory' command output: # Memory used_memory:369616 used_memory_human:360.95K ... """ _, std_out, _ = self.run_redis_command("info memory") for line in std_out.split("\n"): if not line.startswith("used_memory_human"): continue _, used_memory_value = line.split(":") return used_memory_value if used_memory_value != "null" else None Submit