from .common import pp, terminal_can_colors, to_bytes, to_bool, to_str
-from .cfg_app import PpCfgAppError, PpConfigApplication
+from .cfg_app import PpCfgAppError
-__version__ = '0.5.4'
+from .homes_admin import PpHomesAdminError, PpHomesAdminApp
+
+__version__ = '0.6.1'
LOG = logging.getLogger(__name__)
UTC = datetime.timezone.utc
# =============================================================================
-class PpQuotaCheckError(PpCfgAppError):
+class PpQuotaCheckError(PpHomesAdminError):
pass
# =============================================================================
-class PpQuotaCheckApp(PpConfigApplication):
+class PpQuotaCheckApp(PpHomesAdminApp):
"""
Class for the 'quota-check' application to check the utilization
of the home share on the NFS server.
"""
- # /mnt/nfs
- default_chroot_homedir = os.sep + os.path.join('mnt', 'nfs')
- # /home
- default_home_root = os.sep + 'home'
-
default_quota_kb = 300 * 1024
+ # /var/lib/quota-check
default_status_dir = os.sep + os.path.join('var', 'lib', 'quota-check')
+ # /var/lib/quota-check/quota-check.yaml
default_statusfile_base = 'quota-check.yaml'
du_line_re = re.compile(r'^\s*(\d+)\s+(.*)')
# -------------------------------------------------------------------------
def __init__(self, appname=None, version=__version__):
- self.default_mail_recipients = [
- 'admin.berlin@pixelpark.com'
- ]
- self.default_mail_cc = []
-
self.default_reply_to = 'noreply@pixelpark.com'
- self.chroot_homedir = self.default_chroot_homedir
- self.home_root_abs = self.default_home_root
- self.home_root_rel = os.path.relpath(self.home_root_abs, os.sep)
-
self.quota_kb = self.default_quota_kb
self.status_dir = self.default_status_dir
self.du_cmd = self.get_command('du', quiet=True)
self.do_statistics = False
+ self.show_simulate_opt = True
+ self._simulate_opt_help = textwrap.dedent('''\
+ Retrieving all utilization values, but not writing it
+ into th status file.
+ ''').strip()
+
description = textwrap.dedent('''\
This checks the utilization of the home directories on the NFS server
and sends a mail per request about all home directories, which are
help="Quota value in MB (default: {} MB).".format(def_mb),
)
+ self.arg_parser.add_argument(
+ '--status-file',
+ metavar='FILE', dest='status_file',
+ help=("The YAML file containing the statistics of the current week "
+ "(default: {!r}).").format(self.statusfile)
+ )
+
self.arg_parser.add_argument(
'-S', '--stats',
action="store_true", dest="stats",
else:
self.quota_kb = quota
- if 'chroot_homedir' in section:
- v = section['chroot_homedir']
- if not os.path.isabs(v):
- msg = (
- "The chrooted path of the home directories must be an "
- "absolute pathname (found [{s}]/chroot_homedir "
- "=> {v!r} in configuration.").format(s=section_name, v=v)
- raise PpMkHomeError(msg)
- self.chroot_homedir = v
-
- if 'home_root' in section:
- v = section['home_root']
- if not os.path.isabs(v):
- msg = (
- "The root path of the home directories must be an "
- "absolute pathname (found [{s}]/home_root "
- "=> {v!r} in configuration.").format(s=section_name, v=v)
- raise PpMkHomeError(msg)
- self.home_root_abs = v
-
if 'status_file' in section:
v = section['status_file']
if os.path.isabs(v):
self.status_dir = os.path.dirname(self.statusfile)
self.statusfile_base = os.path.basename(self.statusfile)
- self.home_root_rel = os.path.relpath(self.home_root_abs, os.sep)
- self.home_root_real = os.path.join(self.chroot_homedir, self.home_root_rel)
-
cmdline_quota = getattr(self.args, 'quota_mb', None)
if cmdline_quota is not None:
self.quota_kb = cmdline_quota * 1024
+ sfile = getattr(self.args, 'status_file')
+ if sfile:
+ if os.path.isabs(sfile):
+ self.status_dir = os.path.normpath(os.path.dirname(sfile))
+ self.statusfile_base = os.path.basename(sfile)
+ self.statusfile = os.path.normpath(sfile)
+ else:
+ self.statusfile = os.path.normpath(os.path.join(self.status_dir, sfile))
+ self.status_dir = os.path.dirname(self.statusfile)
+ self.statusfile_base = os.path.basename(self.statusfile)
+
self.do_statistics = bool(getattr(self.args, 'stats', False))
# -------------------------------------------------------------------------
self.status_data = self.read_status_data()
self.status_data['last_check'] = self.now
self.read_passwd_data()
- self.check_homes()
+ self.check_home_utilization()
self.write_status_data()
new_fname = "{s}.{d}{e}".format(
s=stem, d=date.strftime('%Y-%m-%d_%H:%M:%S'), e=ext)
LOG.info("Renaming {o!r} -> {n!r}.".format(o=self.statusfile, n=new_fname))
+ if self.simulate:
+ LOG.info("Simulation mode, status file will not be renamed.")
+ return
os.rename(self.statusfile, new_fname)
# -------------------------------------------------------------------------
def compress_old_status_files(self):
+ if self.simulate:
+ LOG.info("Simulation mode, status file rotation will not be executed.")
+ return
+
(stem, ext) = os.path.splitext(self.statusfile_base)
search_base = "{s}.20*{e}".format(s=stem, e=ext)
seach_pattern = os.path.join(self.status_dir, search_base)
if self.verbose > 2:
LOG.debug("Status to write:\n{}".format(pp(self.status_data)))
+ if self.simulate:
+ LOG.info("Simulation mode, status file will not be really written.")
+ return
+
open_args = {}
if six.PY3:
open_args['encoding'] = 'utf-8'
return result
# -------------------------------------------------------------------------
- def check_homes(self):
+ def check_home_utilization(self):
LOG.info("Checking utilization of home directories ...")