Edit File: imunify360-watchdog
#!/opt/imunify360/venv/bin/python3 import argparse import json import logging import logging.handlers import os.path import shutil import socket import subprocess import time from pathlib import Path from typing import Optional from defence360agent import sentry from defence360agent.contracts.config import GENERIC_SENSOR_SOCKET_PATH from defence360agent.utils import is_centos6_or_cloudlinux6 logging.raiseExceptions = False CONNECT_TIMEOUT = 10 REQUEST_TIMEOUT = 60 RETRY_DELAY = 10 MIGRATION_TIMEOUT = 4 * 60 * 60 # 4 hours IMUNIFY360 = "imunify360" IMUNIFY360_AGENT = "imunify360-agent" IMUNIFY360_AGENT_SOCKET = "imunify360-agent.socket" AGENT_SOCKET_PATH = "/var/run/defence360agent/simple_rpc.sock" SERVICE = "service" SUBPROCESS_TIMEOUT = 1800 RESTART = "restart" STATUS = "status" SHOW = "show" AGENT_IN_MIGRATION_STATE = "Applying database migrations" I360_GW_DIR = "/var/imunify360/gw.dir" I360_GW_TTL_SEC = 2 * 60 * 60 def run(cmd, *, timeout=SUBPROCESS_TIMEOUT, check=False, **kwargs): """Run *cmd* with *timeout* without raising TimeoutExpired. On timeout, return CompletedProcess with returncode equal to None. """ try: return subprocess.run(cmd, timeout=timeout, check=check, **kwargs) except subprocess.TimeoutExpired as e: return subprocess.CompletedProcess( e.cmd, returncode=None, stdout=e.stdout, stderr=e.stderr ) def service_is_running(systemctl_exec: Optional[str], name: str) -> bool: """Check with help of [systemctl|service] command status of service""" if systemctl_exec: cmd = [systemctl_exec, STATUS, name] else: cmd = [SERVICE, name, STATUS] cp = run( cmd, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) return cp.returncode == 0 def restart_service(systemctl_exec: Optional[str], name: str) -> None: """Check with help of [systemctl|service] command status of service""" if systemctl_exec: cmd = [systemctl_exec, RESTART, name] else: cmd = [SERVICE, name, RESTART] run( cmd, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) def restart_imunify360(systemctl_exec: Optional[str]) -> None: """Restart resident imunify360 services""" restart_service(systemctl_exec, IMUNIFY360) def restart_imunify360_agent(systemctl_exec: Optional[str]) -> None: """Restart non-resident imunify360 services""" restart_service(systemctl_exec, IMUNIFY360_AGENT) def setup_logging(level) -> logging.Logger: logger = logging.getLogger("imunify360-watchdog") logger.setLevel(level) handler = logging.handlers.SysLogHandler("/dev/log") formatter = logging.Formatter("%(name)s: %(message)s") handler.formatter = formatter logger.addHandler(handler) sentry.configure_sentry() return logger def send_to_generic_socket(_msg): with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock: sock.settimeout(CONNECT_TIMEOUT) sock.connect(GENERIC_SENSOR_SOCKET_PATH) msg = json.dumps(_msg).encode() + b"\n" start_time = time.monotonic() sock.settimeout(REQUEST_TIMEOUT) sock.sendall(msg) remaining_time = start_time + REQUEST_TIMEOUT - time.monotonic() if remaining_time <= 0: raise socket.timeout() sock.settimeout(remaining_time) with sock.makefile(encoding="utf-8") as file: response = file.readline() if not response: raise ValueError("Empty response from socket") return json.loads(response) def check_agent_socket_alive(systemctl_exec: Optional[str]) -> bool: if is_centos6_or_cloudlinux6(): return service_is_running(systemctl_exec, IMUNIFY360_AGENT) else: return service_is_running(systemctl_exec, IMUNIFY360_AGENT_SOCKET) def check_outdated_gw_logs(gw_dir: Path, ttl_sec: float) -> bool: mtime_threshold = time.time() - ttl_sec if not gw_dir.exists(): return False files = [] try: for file in gw_dir.iterdir(): if file.suffix == '.log' and file.is_file() and file.stat().st_mtime < mtime_threshold: files.append(file.name) file.unlink() except Exception as e: logging.error("error iterating through log files: %s", e) shutil.rmtree(gw_dir) return False if files: logging.error("outdated files found: %s", files) return files == [] def generic_sensor_with_retries(rpc_timeout: int) -> Optional[dict]: start = time.time() while True: try: return send_to_generic_socket( { "method": "HEALTH", } ) except Exception: if time.time() - start >= rpc_timeout: raise time.sleep(RETRY_DELAY) def systemctl_executable() -> Optional[str]: """Try to find systemctl in default PATH and return None if failed.""" return shutil.which("systemctl", path=os.defpath) def service_is_migrating(systemctl_exec, name, logger): """ Check that service in "apply migrations" state and do not exhaust timeout """ if systemctl_exec: cmd = [ systemctl_exec, SHOW, name, "-p", "StatusText", "-p", "ExecMainStartTimestampMonotonic", ] else: cmd = [SERVICE, name, SHOW] cp = run( cmd, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, ) # Parse only main parameters from output, other lines ignored params = { key: value for (key, value) in [ key_value.split("=", 1) for key_value in cp.stdout.decode().splitlines() ] if key in ["StatusText", "ExecMainStartTimestampMonotonic"] } if AGENT_IN_MIGRATION_STATE in params["StatusText"]: migration_duration = ( time.monotonic() - int(params["ExecMainStartTimestampMonotonic"]) / 1e6 ) logger.info("%s migrating for %d sec", name, migration_duration) if migration_duration < MIGRATION_TIMEOUT: return True logger.error("Migration took too long") return False def ensure_resident_health( logger: logging.Logger, systemctl_exec: Optional[str], rpc_timeout: int ) -> None: try: response = generic_sensor_with_retries(rpc_timeout) except Exception: logger.exception("Restarting resident service due to RPC failures") restart_imunify360(systemctl_exec) return if not response.get("healthy", False): logger.error( "Restarting resident service due to health report: %s", response.get("why") if response.get("why") else response.get("error"), ) restart_imunify360(systemctl_exec) return if not check_outdated_gw_logs(Path(I360_GW_DIR), I360_GW_TTL_SEC): logger.error("Restarting resident service due to outdated files present") restart_imunify360(systemctl_exec) return logger.info("%s is healthy: %s", IMUNIFY360, response.get("why")) def ensure_agent_health( logger: logging.Logger, systemctl_exec: Optional[str] ) -> None: try: # since `service *.sock status` returns 0 even socket # file isn't exists we need to check it manually if ( not check_agent_socket_alive(systemctl_exec) or not Path(AGENT_SOCKET_PATH).exists() ): logger.exception("Restarting agent due to socket failures") restart_imunify360_agent(systemctl_exec) except Exception as e: logger.exception("Restarting agent due to %s", e) restart_imunify360_agent(systemctl_exec) def main(rpc_timeout, log_level=logging.INFO): logger = setup_logging(log_level) systemctl_exec = systemctl_executable() if not service_is_running(systemctl_exec, IMUNIFY360): logger.info("%s is not running", IMUNIFY360) return elif service_is_migrating(systemctl_exec, IMUNIFY360, logger): logger.info("%s is migrating at the moment", IMUNIFY360) return ensure_agent_health(logger, systemctl_exec) ensure_resident_health(logger, systemctl_exec, rpc_timeout) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("rpc_timeout", type=int) return parser.parse_args() if __name__ == "__main__": args = parse_args() main(rpc_timeout=args.rpc_timeout)