Emit a warning to stderr when your disk usage is too high. Great when used from cron. This is a mirror.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

223 lines
4.9 KiB

  1. #!/usr/bin/env python3
  2. """
  3. Mike's Disk Usage Warner
  4. A simple script to emit warnings out to stderr if a disk's usage surpasses a threshold
  5. Copyright 2019 Mike Peralta; All rights reserved
  6. Released under the GNU GENERAL PUBLIC LICENSE v3 (See LICENSE file for more)
  7. """
  8. #
  9. import os
  10. import re
  11. import subprocess
  12. import sys
  13. import yaml
  14. #
  15. class DiskUsageWarn:
  16. def __init__(self):
  17. self.__config_paths = []
  18. self.__configs = []
  19. self.consume_arguments()
  20. def log(self, s, o=None):
  21. message = "[Disk-Usage-Warn] " + s
  22. if o:
  23. message += " " + str(o)
  24. print(message)
  25. self.execute_command(["logger", message])
  26. def stderr(self, s, o=None):
  27. message = "[Disk-Usage-Warn] " + s
  28. if o:
  29. message += " " + str(o)
  30. print(message, file=sys.stderr)
  31. self.execute_command(["logger", message])
  32. def consume_arguments(self):
  33. self.__config_paths = []
  34. for i in range(1, len(sys.argv)):
  35. arg = sys.argv[i]
  36. if arg == "--config":
  37. i, one_path = self.consume_argument_companion(i)
  38. self.__config_paths.append(one_path)
  39. self.log("Found config path argument:", one_path)
  40. @staticmethod
  41. def consume_argument_companion(arg_index):
  42. companion_index = arg_index + 1
  43. if companion_index >= len(sys.argv):
  44. raise Exception("Expected argument after", sys.argv[arg_index])
  45. return companion_index, sys.argv[companion_index]
  46. #
  47. def consume_configs(self):
  48. #
  49. self.__configs = []
  50. #
  51. for path in self.__config_paths:
  52. if os.path.isdir(path):
  53. self.log(path + " is actually a directory; Iterating over contained files (non-recursive)")
  54. for file_name in os.listdir(path):
  55. one_config_path = os.path.join(path, file_name)
  56. if re.match(".+\.yaml$", file_name):
  57. self.log("Found yaml: " + file_name)
  58. self.consume_config(one_config_path)
  59. else:
  60. self.log("Ignoring non-yaml file: " + file_name)
  61. elif os.path.isfile(path):
  62. self.consume_config(path)
  63. else:
  64. raise Exception("Don't know what to do with config path:" + str(path))
  65. self.log("Consumed " + str(len(self.__configs)) + " configs")
  66. #
  67. def consume_config(self, path: str):
  68. config = self.load_config(path)
  69. self.__configs.append(config)
  70. self.log("Consumed config: " + path)
  71. @staticmethod
  72. def load_config(path: str):
  73. # Open the file
  74. f = open(path)
  75. if not f:
  76. raise Exception("Unable to open config file: " + path)
  77. # Parse
  78. config = yaml.safe_load(f)
  79. # Add the config file's own path
  80. config["path"] = path
  81. return config
  82. def run(self):
  83. self.consume_configs()
  84. self.do_configs()
  85. def do_configs(self):
  86. for config in self.__configs:
  87. self.do_config(config)
  88. def do_config(self, config):
  89. # Pull the max usage
  90. self.assert_config_key(config, "max-usage", [int, str, list])
  91. max_percent = str(config["max-usage"])
  92. match = re.match("(?P<integer_percent>[0-9]+)%?", max_percent)
  93. if not match:
  94. raise Exception("Unable to parse configuration value for max-usage (integer percent)")
  95. max_percent = int(match.group("integer_percent"))
  96. # Check each device
  97. self.assert_config_key(config, "devices", list)
  98. for device in config["devices"]:
  99. device_usage = self.get_device_usage(device)
  100. self.log("Device Usage: " + device + " ==> " + str(device_usage) + "%")
  101. if device_usage > max_percent:
  102. error_message = ("Device " + str(device) + " is too full"
  103. + "; Using " + str(device_usage) + "%"
  104. + ", but maximum is " + str(max_percent) + "%"
  105. )
  106. self.stderr(error_message)
  107. @staticmethod
  108. def assert_config_key(config, key, types):
  109. if not isinstance(types, list):
  110. types = [types]
  111. if len(types) == 0:
  112. types = [list]
  113. if key not in config.keys():
  114. raise Exception("Missing required config key: " + str(key))
  115. for t in types:
  116. if isinstance(config[key], t):
  117. return True
  118. raise Exception("Config key \"" + key + "\" should be one of these types \"" + str(types) + "\"")
  119. def get_device_usage(self, device):
  120. args = ["df", device]
  121. returncode, stdout, stderr = self.execute_command(args)
  122. if returncode != 0:
  123. raise Exception("Failed to poll device usage\n" + stderr)
  124. # Grab percent
  125. pattern = re.compile(".*?(?P<percent_integer>[0-9]+)%.*?", re.DOTALL)
  126. match = pattern.match(str(stdout))
  127. if not match:
  128. raise Exception("Unable to parse device usage from:\n" + str(stdout))
  129. percent_integer = int(match.group("percent_integer"))
  130. return percent_integer
  131. @staticmethod
  132. def execute_command(args):
  133. # Start the process
  134. #process = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
  135. process = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  136. #
  137. stdout, stderr = process.communicate()
  138. if stdout:
  139. stdout = stdout.decode()
  140. if stderr:
  141. stderr = stderr.decode()
  142. return process.returncode, stdout, stderr
  143. def main():
  144. duw = DiskUsageWarn()
  145. duw.run()
  146. if __name__ == "__main__":
  147. main()