""" Monitors memory usage, notifies and stops/kills the top memory usage process. Test with: stress --vm-bytes 7G --vm-hang 0 --vm 1 """ import argparse import logging import time import config import lib logging.basicConfig() logger = logging.getLogger() def setup_logging_level(debug=False): log_level = logging.DEBUG if debug else logging.ERROR logger.setLevel(log_level) logger.debug("Debugging enabled") def parse_args(): parser = argparse.ArgumentParser() parser.add_argument('path', help="Path to monitor") parser.add_argument('--debug', dest='debug', action='store_true') return parser.parse_args() def get_available_mem_g(): mem_available = lib.run_os_command( "grep MemAvailable /proc/meminfo | awk '{print $2}'") mem_available = int(mem_available) mem_available_g = round(mem_available / (1024 ** 2), 1) return mem_available_g def get_top_pid_comm(): top_pid_comm = lib.run_os_command( "ps -eo pid,pmem,comm --sort=-pmem | head -n 2 | awk '{print $1 \" \" $3}' | tail -1") return top_pid_comm.split() def stop_process(pid): lib.run_os_command(f"kill -{config.SIGNAL} {pid}") def main(): is_low_mem = False while True: mem_available_g = get_available_mem_g() logging.debug(f"Memory Available: {mem_available_g} G") top_pid, top_comm = get_top_pid_comm() if mem_available_g < config.MEM_MIN_AVAILABLE_G: if not is_low_mem: message = f"Less than {config.MEM_MIN_AVAILABLE_G}G of memory available. \n\n Sent signal {config.SIGNAL} {top_comm} ({top_pid}) {config.STR_CONT_OPT} \n\n Run 'kill -SIGKILL ' to kill immediately." logging.warning(message) stop_process(int(top_pid)) lib.send_notification("Memory Alert", message) is_low_mem = True else: is_low_mem = False logging.debug( f"Top process by memory usage: PID {top_pid}, Command: {top_comm}") time.sleep(config.TIME_SLEEP_MEM_S) if __name__ == "__main__": main()