Gam

??? 123123123123 ..................................................................................................................................... Gam

File manager - Edit - /opt/cloudlinux/venv/lib/python3.11/site-packages/clconfig/lve_stats2_reseller_lib.py

Back
# -- coding: utf-8 -- # lve_stats2_reseller_lib.py - lve-stats reseller 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.utils import mod_makedirs from . import lve_stats2_lib 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, ) from .lve_stats2_lib import LveStats2Exception, StatsNotifierConfig, StatsNotifierDir, faults_dict ResellerStatsNotifierDir = StatsNotifierDir + '/resellers.d/' reseller_notify_dict = { 'NOTIFY_RESELLER_ON_TOTAL_FAULTS': 'notifyReseller', 'NOTIFY_RESELLER_ON_CUSTOMERS_FAULTS': 'notifyResellerOnCustomers', 'NOTIFY_CUSTOMERS_ON_FAULTS': 'notifyCustomers' } reseller_number_notify_dict = { 'NOTIFY_MIN_FAULTS_RESELLER': 'reseller', 'NOTIFY_MIN_FAULTS_CUSTOMER': 'customer' } reseller_notify_time_dict = { 'NOTIFY_INTERVAL_RESELLER': 'reseller', 'NOTIFY_INTERVAL_CUSTOMER': 'customer' } # For unit tests def _open(file_name, mode='r'): return open(file_name, mode=mode, encoding='utf-8') def _create_default_reseller_config(): """ Creates default StatsNotifier config for reseller """ admin_config = lve_stats2_lib.get_notification().get('faultsNotification') admin_faults = {key: admin_config['faultsToInclude'][key] for key in ( 'mem', 'iops', 'io', 'nproc', 'concurrentConnections', 'cpu' )} return {'faultsNotification': {'notifyResellerOnCustomers': admin_config['notifyResellers'], 'minimumNumberOfFaultsToNotify': { 'reseller': admin_config['minimumNumberOfFaultsToNotify']['admin'], 'customer': admin_config['minimumNumberOfFaultsToNotify']['user']}, 'notifyReseller': admin_config['notifyResellers'], 'notify': {'reseller': {'unitOfTime': admin_config['notify']['admin']['unitOfTime'], 'period': admin_config['notify']['admin']['period']}, 'customer': {'unitOfTime': admin_config['notify']['user']['unitOfTime'], 'period': admin_config['notify']['user']['period']} }, 'faultsToInclude': admin_faults, 'notifyCustomers': admin_config['notifyResellerCustomers'] }} def get_notification(reseller, defaults=False): """ Retrieves lve-stats2 notifications parameters for resellers :param bool defaults: True if we want get defaults when config is not set, else False :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 } } """ if reseller is None: raise ValueError('reseller can`t be None') config_path = os.path.join(ResellerStatsNotifierDir, f"{reseller}_StatsNotifier.cfg") if not os.path.exists(config_path): return _create_default_reseller_config() if defaults else {} 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: if line.startswith('#') or line.strip() == '': continue key, value = line.split('=') config_dict[key] = value.strip() notify_d = reseller_notify_dict faults_d = faults_dict number_notify_d = reseller_number_notify_dict notify_time_d = reseller_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) faults_notification_dict['faultsToInclude'] = faults_include_dict faults_notification_dict['minimumNumberOfFaultsToNotify'] = min_num_notify faults_notification_dict['notify'] = period_notify return {"faultsNotification": faults_notification_dict} def set_notification(parameters_dict, reseller): """ 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 = {} if reseller is None: raise ValueError('reseller can`t be None') config_path = os.path.join(ResellerStatsNotifierDir, f"{reseller}_StatsNotifier.cfg") if not os.path.exists(config_path): notifier_params.update(_create_default_reseller_config()) else: notifier_params.update(get_notification(reseller=reseller)) # 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_RESELLER': (lambda x, y: f"{is_positive_int(x)}{time_unit_to_letter(y)}", 'notify/reseller/period', 'notify/reseller/unitOfTime'), 'NOTIFY_INTERVAL_CUSTOMER': (lambda x, y: f"{is_positive_int(x)}{time_unit_to_letter(y)}", 'notify/customer/period', 'notify/customer/unitOfTime'), 'NOTIFY_RESELLER_ON_TOTAL_FAULTS': (lambda x: boolean_to_yes_no(x), 'notifyReseller'), 'NOTIFY_RESELLER_ON_CUSTOMERS_FAULTS': (lambda x: boolean_to_yes_no(x), 'notifyResellerOnCustomers'), 'NOTIFY_CUSTOMERS_ON_FAULTS': (lambda x: boolean_to_yes_no(x), 'notifyCustomers'), 'NOTIFY_MIN_FAULTS_RESELLER': 'minimumNumberOfFaultsToNotify/reseller', 'NOTIFY_MIN_FAULTS_CUSTOMER': 'minimumNumberOfFaultsToNotify/customer'}) 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 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}") # 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)