[ PARENT ]

EDIT :: lve_stats2_lib.py

# -*- coding: utf-8 -*- # lve_stats2_lib.py - lve-stats library for cloudlinux-config library # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT import os import subprocess from clcommon.clexception import FormattedException from clcommon.utils import mod_makedirs from .clconfig_utils import ( boolean_to_yes_no, is_positive_int, min_num_notify_converter, repack_dict, str_to_boolean, time_convertor_to_dict, time_unit_to_letter, ) class LveStats2Exception(FormattedException): pass StatsNotifierDir = '/etc/sysconfig/lvestats.config' StatsNotifierConfig = StatsNotifierDir + '/StatsNotifier.cfg' notify_dict = {'NOTIFY_ADMIN': 'notifyAdmin', 'NOTIFY_RESELLER': 'notifyResellers', 'NOTIFY_CUSTOMER': 'notifyCustomers', 'NOTIFY_INCLUDE_RESELLER_CUSTOMER': 'notifyResellerCustomers'} faults_dict = {'NOTIFY_MEMORY': 'mem', 'NOTIFY_IOPS': 'iops', 'NOTIFY_NPROC': 'nproc', 'NOTIFY_IO': 'io', 'NOTIFY_EP': 'concurrentConnections', 'NOTIFY_CPU': 'cpu'} email_notify_dict = {'NOTIFY_FROM_EMAIL': 'notifyFromEmail', 'REPORT_ADMIN_EMAIL': 'reportAdminMail', 'NOTIFY_FROM_SUBJECT': 'notifySubject', 'NOTIFY_CHARSET_EMAIL': 'notifyCharset'} number_notify_dict = {'NOTIFY_MIN_FAULTS_USER': 'user', 'NOTIFY_MIN_FAULTS_ADMIN': 'admin'} notify_time_dict = {'NOTIFY_INTERVAL_ADMIN': 'admin', 'NOTIFY_INTERVAL_USER': 'user'} # For unit tests def _open(file_name, mode='r'): return open(file_name, mode=mode, encoding='utf-8') def get_notification(reseller=None): """ Retrieves lve-stats2 notifications parameters :param str reseller: reseller name, if we want notifications parameters for a reseller :return: dict. For example: {'faultsNotification': {'notifyResellerCustomers': True, 'notifyResellers': True, 'minimumNumberOfFaultsToNotify': {'admin': 0, 'user': 0}, 'notifyAdmin': True, 'notify': {'admin': {'unitOfTime': 'hours', 'period': 12}, 'user': {'unitOfTime': 'hours', 'period': 12} }, 'faultsToInclude': {'mem': True, 'iops': False, 'io': False, 'nproc': False, 'concurrentConnections': False, 'cpu': True}, 'notifyCustomers': False } } """ config_path = StatsNotifierConfig if not os.path.exists(config_path): return {} try: f = _open(config_path) config = f.readlines() f.close() except (IOError, OSError) as e: raise LveStats2Exception({'message': "%(lvestats_cfg)s read error: " + str(e), 'context': {'lvestats_cfg': config_path}}) from e config_dict = {} for line in config: # skip comments and empty lines if line.startswith('#') or line.strip() == '': continue key, value = line.split('=') config_dict[key] = value.strip() notify_d = notify_dict faults_d = faults_dict number_notify_d = number_notify_dict notify_time_d = notify_time_dict faults_notification_dict = repack_dict(notify_d, config_dict, str_to_boolean) faults_include_dict = repack_dict(faults_d, config_dict, str_to_boolean) min_num_notify = repack_dict(number_notify_d, config_dict, min_num_notify_converter, default=1) period_notify = repack_dict(notify_time_d, config_dict, time_convertor_to_dict) email_settings = repack_dict(email_notify_dict, config_dict, str, default=None) faults_notification_dict['faultsToInclude'] = faults_include_dict faults_notification_dict['minimumNumberOfFaultsToNotify'] = min_num_notify faults_notification_dict['notify'] = period_notify faults_notification_dict['email'] = email_settings return {"faultsNotification": faults_notification_dict} def set_notification(parameters_dict, reseller=None): """ Sets lve-stats2 notifications parameters :param parameters_dict: Parametres to set. For example: {u'notifyResellers': True, u'faultsToInclude': {u'mem': True, u'iops': False, u'io': False, u'nproc': False, u'concurrentConnections': False, u'cpu': True}, u'minimumNumberOfFaultsToNotify': {u'admin': 0, u'user': 0}, u'notifyAdmin': True, u'notify': {u'admin': {u'unitOfTime': u'hours', u'period': 12}, u'user': {u'unitOfTime': u'hours', u'period': 12} }, u'notifyResellerCustomers': True, u'notifyCustomers': False } :param str reseller: either reseller name or None """ # Do nothing, if admin's lvestats2 notifier config not found # either parameters_dict is empty and we shouldn't change anything if not os.path.exists(StatsNotifierConfig) or len(parameters_dict) == 0: return notifier_params = {} config_path = StatsNotifierConfig notifier_params = get_notification() # create directory and empty config if not exists if not os.path.exists(os.path.dirname(config_path)): mod_makedirs(os.path.dirname(config_path), 0o755) if not os.path.exists(config_path): _open(config_path, 'w').close() # Read notifier config try: f = _open(config_path) config_lines = [line.strip() for line in f.readlines()] f.close() except (IOError, OSError) as e: raise LveStats2Exception({'message': "%(lvestats_cfg)s read error: " + str(e), 'context': {'lvestats_cfg': StatsNotifierConfig}}) from e # Fill parameters to change lve-stats config faults_config_map = {'NOTIFY_CPU': (lambda x: boolean_to_yes_no(x), 'faultsToInclude/cpu'), 'NOTIFY_NPROC': (lambda x: boolean_to_yes_no(x), 'faultsToInclude/nproc'), 'NOTIFY_IO': (lambda x: boolean_to_yes_no(x), 'faultsToInclude/io'), 'NOTIFY_MEMORY': (lambda x: boolean_to_yes_no(x), 'faultsToInclude/mem'), 'NOTIFY_EP': (lambda x: boolean_to_yes_no(x), 'faultsToInclude/concurrentConnections'), 'NOTIFY_IOPS': (lambda x: boolean_to_yes_no(x), 'faultsToInclude/iops')} faults_config_map.update({ 'NOTIFY_INTERVAL_ADMIN': (lambda x, y: f"{is_positive_int(x)}{time_unit_to_letter(y)}", 'notify/admin/period', 'notify/admin/unitOfTime'), 'NOTIFY_INTERVAL_USER': (lambda x, y: f"{is_positive_int(x)}{time_unit_to_letter(y)}", 'notify/user/period', 'notify/user/unitOfTime'), 'NOTIFY_ADMIN': (lambda x: boolean_to_yes_no(x), 'notifyAdmin'), 'NOTIFY_CUSTOMER': (lambda x: boolean_to_yes_no(x), 'notifyCustomers'), 'NOTIFY_RESELLER': (lambda x: boolean_to_yes_no(x), 'notifyResellers'), 'NOTIFY_INCLUDE_RESELLER_CUSTOMER': (lambda x: boolean_to_yes_no(x), 'notifyResellerCustomers'), 'NOTIFY_MIN_FAULTS_ADMIN': 'minimumNumberOfFaultsToNotify/admin', 'NOTIFY_MIN_FAULTS_USER': 'minimumNumberOfFaultsToNotify/user'}) # Update dict with email parameters # If None is passed, line will be removed from config before writing it back to file faults_config_map.update({ 'NOTIFY_FROM_EMAIL': 'email/notifyFromEmail', 'NOTIFY_FROM_SUBJECT': 'email/notifySubject', 'REPORT_ADMIN_EMAIL': 'email/reportAdminMail', 'NOTIFY_CHARSET_EMAIL': 'email/notifyCharset'}) new_config = {} def get_val_by_path(path): path_parts = path.split('/') try: point = parameters_dict for part in path_parts: point = point[part] except KeyError: point = notifier_params['faultsNotification'] for part in path_parts: point = point[part] return point for key, val in faults_config_map.items(): try: if isinstance(val, str): new_config[key] = get_val_by_path(val) else: new_config[key] = val[0](*(get_val_by_path(x) for x in val[1:])) except KeyError: continue # to do not change iter object while iterate it config_to_write = [line.strip() for line in config_lines] try: for idx, line in enumerate(config_lines): # Pass comment and empty lines if line.startswith('#') or line.strip() == '': continue try: key, value = line.split('=') except ValueError as e: raise LveStats2Exception({ 'message': '%(lvestats_cfg)s format error: ' + f'{idx + 1}: {line}', 'context': {'lvestats_cfg': config_path}}) from e if key in new_config: # Parameter must be changed config_to_write[idx] = f"{key}={new_config[key]}" # Remove used parameter del new_config[key] except (KeyError, IndexError) as e: raise LveStats2Exception({'message': "%(lvestats_cfg)s format error: " + str(e), 'context': {'lvestats_cfg': config_path}}) from e # Add absent options for key, value in new_config.items(): config_to_write.append(f"{key}={value}") # Email notifications are optional # They should be removed from config, if None value is passed def filter_config_lines(_line): if _line.startswith('#') or _line.strip() == '': return True _, _value = _line.split('=') return _value != 'None' config_to_write = list(filter(filter_config_lines, config_to_write)) # Write config back to file try: with open(config_path, 'w', encoding='utf-8') as f: f.writelines('\n'.join(config_to_write)) except (IOError, OSError) as e: raise LveStats2Exception({'message': "%(lvestats_cfg)s write error: " + str(e), 'context': {'lvestats_cfg': config_path}}) from e # Restart lvestats server # /sbin/service lvestats reload subprocess.run('/sbin/service lvestats reload &>/dev/null &', shell=True, executable='/bin/bash', check=False)

SAVE CANCEL